eta_index_lib/models/base_from_mysteel_chemical.go

package models

import (
	"eta/eta_index_lib/utils"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/beego/beego/v2/client/orm"
)

//钢联

type MysteelChemicalData struct {
	InputValue float64 `orm:"column(value)" description:"值"`
	DataTime   string  `orm:"column(data_time)" description:"日期"`
}

func GetMysteelChemicalDataByCondition(condition string, pars []interface{}) (item []*MysteelChemicalData, err error) {
	sql1 := ` SELECT * FROM base_from_mysteel_chemical_data WHERE 1=1  `
	o := orm.NewOrm()
	if condition != "" {
		sql1 += condition
	}
	sql := `select * from (` + sql1 + ` having 1 order by modify_time DESC ) tmp GROUP BY data_time ORDER BY data_time DESC `
	_, err = o.Raw(sql, pars).QueryRows(&item)
	return
}

// AddEdbDataFromMysteelChemical 新增钢联指标数据
func AddEdbDataFromMysteelChemical(edbCode string) (err error) {
	o := orm.NewOrm()

	var condition string
	var pars []interface{}

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

	mysteelChemicalDataList, err := GetMysteelChemicalDataByCondition(condition, pars)
	if err != nil {
		return
	}

	dataLen := len(mysteelChemicalDataList)

	existMap := make(map[string]string)
	if dataLen > 0 {
		var isAdd bool
		addSql := ` INSERT INTO edb_data_mysteel_chemical (edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
		for i := 0; i < dataLen; i++ {
			item := mysteelChemicalDataList[i]
			eDate := item.DataTime
			sValue := utils.SubFloatToString(item.InputValue, 30)
			if sValue != "" {
				if _, ok := existMap[eDate]; !ok {
					dataTime, err := time.ParseInLocation(utils.FormatDate, eDate, time.Local)
					if err != nil {
						return err
					}
					timestamp := dataTime.UnixNano() / 1e6
					timeStr := fmt.Sprintf("%d", timestamp)
					addSql += GetAddSql("0", edbCode, eDate, timeStr, sValue)
					isAdd = true
				}
			}
			existMap[eDate] = eDate
		}
		if isAdd {
			addSql = strings.TrimRight(addSql, ",")
			utils.FileLog.Info("addSql:" + addSql)
			_, err = o.Raw(addSql).Exec()
			if err != nil {
				return err
			}
		}
	}
	return
}

// RefreshEdbDataFromMysteelChemical 刷新钢联指标数据
func RefreshEdbDataFromMysteelChemical(edbInfoId int, edbCode, startDate string) (err error) {
	source := utils.DATA_SOURCE_MYSTEEL_CHEMICAL
	subSource := utils.DATA_SUB_SOURCE_EDB

	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)
	}

	mysteelChemicalDataList, err := GetMysteelChemicalDataByCondition(condition, pars)
	if err != nil {
		return
	}

	// 真实数据的最大日期  ， 插入规则配置的日期
	var realDataMaxDate, edbDataInsertConfigDate time.Time
	var edbDataInsertConfig *EdbDataInsertConfig
	var isFindConfigDateRealData bool //是否找到配置日期的实际数据的值
	{
		edbDataInsertConfig, err = GetEdbDataInsertConfigByEdbId(edbInfoId)
		if err != nil && err.Error() != utils.ErrNoRow() {
			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, subSource, 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_mysteel_chemical(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
	var isAdd bool
	addMap := make(map[string]string)
	for _, v := range mysteelChemicalDataList {
		item := v
		eDate := item.DataTime
		sValue := utils.SubFloatToString(item.InputValue, 30)

		dataTime, err := time.ParseInLocation(utils.FormatDate, eDate, time.Local)
		if err != nil {
			return err
		}
		if findItem, ok := existMap[v.DataTime]; !ok {
			if sValue != "" {
				timestamp := dataTime.UnixNano() / 1e6
				timeStr := fmt.Sprintf("%d", timestamp)
				saveValue := sValue

				if _, addOk := addMap[eDate]; !addOk {
					addSql += GetAddSql(edbInfoIdStr, edbCode, eDate, timeStr, saveValue)
					isAdd = true
				}
			}
		} else {
			if findItem != nil && utils.SubFloatToString(findItem.Value, 30) != sValue {
				err = ModifyEdbDataById(source, subSource, findItem.EdbDataId, sValue)
				if err != nil {
					return err
				}
			}
		}
		addMap[v.DataTime] = v.DataTime

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

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

	if isAdd {
		addSql = strings.TrimRight(addSql, ",")
		_, err = o.Raw(addSql).Exec()
		if err != nil {
			return err
		}
	}
	return
}

// 钢联化工指标数据
type BaseFromMysteelChemicalIndex struct {
	BaseFromMysteelChemicalIndexId    int64     `orm:"column(base_from_mysteel_chemical_index_id);pk"`
	BaseFromMysteelChemicalClassifyId int       `description:"分类id"`
	IndexCode                         string    `description:"指标编码"`
	IndexName                         string    `description:"指标名称"`
	Unit                              string    `description:"单位"`
	Source                            string    `description:"数据来源"`
	Frequency                         string    `description:"频度"`
	StartDate                         time.Time `description:"开始日期"`
	EndDate                           time.Time `description:"结束日期"`
	Describe                          string    `description:"指标描述"`
	UpdateWeek                        string    `description:"更新周期"`
	UpdateTime                        string    `description:"更新时间，多个时间点用英文,隔开"`
	UpdateTime2                       string    `description:"更新时间2"`
	SysUserId                         int       `description:"创建人id"`
	SysUserRealName                   string    `description:"创建人姓名"`
	FilePath                          string    `description:"文件存储路径"`
	MergeFilePath                     string    `description:"更新文件"`
	FileIndex                         int       `description:"文件索引"`
	MergeUpdateWeek                   string    `description:"合并文件的更新周"`
	UpdateDate                        string    `description:"更新日期"`
	CreateTime                        time.Time `description:"创建时间"`
	ModifyTime                        time.Time `description:"修改时间"`
	IsStop                            int       `description:"是否停更:1:停更,0:未停更"`
	TerminalCode                      string    `description:"终端编码"`
}

// GetIndexRefreshAllByMergeFile 根据合并文件去分组查询需要刷新的文件
func (m *BaseFromMysteelChemicalIndex) GetIndexRefreshAllByMergeFile() (items []*BaseFromMysteelChemicalIndex, err error) {
	o := orm.NewOrm()
	sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE merge_file_path <>'' AND base_from_mysteel_chemical_classify_id NOT IN(54,55,56,57) GROUP BY merge_file_path`
	_, err = o.Raw(sql).QueryRows(&items)
	return
}

func (m *BaseFromMysteelChemicalIndex) GetIndexItem(indexCode string) (item *BaseFromMysteelChemicalIndex, err error) {
	o := orm.NewOrm()
	sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE index_code = ? `
	err = o.Raw(sql, indexCode).QueryRow(&item)
	return
}

func (m *BaseFromMysteelChemicalIndex) GetIndexCreate(terminalCode string) (items []*BaseFromMysteelChemicalIndex, err error) {
	o := orm.NewOrm()
	sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE index_name = '' AND terminal_code = ? `
	_, err = o.Raw(sql, terminalCode).QueryRows(&items)
	return
}

// Add 新增
func (m *BaseFromMysteelChemicalIndex) Add() (err error) {
	o := orm.NewOrm()
	_, err = o.Insert(m)
	return
}

func (m *BaseFromMysteelChemicalIndex) Update(cols []string) (err error) {
	o := orm.NewOrm()
	_, err = o.Update(m, cols...)
	return
}

// GetNoMergeIndexByFrequencyCount 获取未合并的指标总数量
func (m *BaseFromMysteelChemicalIndex) GetNoMergeIndexByFrequencyCount(frequency string) (total int64, err error) {
	o := orm.NewOrm()
	sql := `SELECT count(1) AS total FROM base_from_mysteel_chemical_index WHERE frequency = ? AND merge_file_path = '' AND index_name NOT LIKE "%停%"  `
	err = o.Raw(sql, frequency).QueryRow(&total)

	return
}

// GetMaxFileIndexReq 获取最大文件数请求
type GetMaxFileIndexReq struct {
	Frequency    string
	EndDate      string
	TerminalCode string `description:"终端编码"`
}

// GetMaxFileIndex 获取最大文件的
func (m *BaseFromMysteelChemicalIndex) GetMaxFileIndex(frequency string) (item *BaseFromMysteelChemicalIndex, err error) {
	o := orm.NewOrm()
	sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE frequency=? AND index_name NOT LIKE "%停%" ORDER BY file_index DESC `
	err = o.Raw(sql, frequency).QueryRow(&item)

	return
}

// GetNoMergeIndexByFrequencyPageReq 获取未合并的指标列表
type GetNoMergeIndexByFrequencyPageReq struct {
	Frequency string
	Limit     int
}

// GetNoMergeIndexByFrequencyPage 获取未合并的指标列表
func (m *BaseFromMysteelChemicalIndex) GetNoMergeIndexByFrequencyPage(frequency string, limit int) (items []*BaseFromMysteelChemicalIndex, err error) {
	o := orm.NewOrm()
	sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE frequency=? AND merge_file_path = '' limit ? `
	_, err = o.Raw(sql, frequency, limit).QueryRows(&items)
	return
}

type GetIndexCountByMergeFilePathReq struct {
	MergeFilePath string
}

func (m *BaseFromMysteelChemicalIndex) GetIndexCountByMergeFilePath(mergeFilePath string) (total int64, err error) {
	o := orm.NewOrm()
	sql := `SELECT count(1) AS total FROM base_from_mysteel_chemical_index WHERE merge_file_path = ?`
	err = o.Raw(sql, mergeFilePath).QueryRow(&total)
	return
}

// GetIndexRefreshMethanolByMergeFile 根据合并文件去分组查询需要刷新的甲醇文件
func (m *BaseFromMysteelChemicalIndex) GetIndexRefreshMethanolByMergeFile() (items []*BaseFromMysteelChemicalIndex, err error) {
	o := orm.NewOrm()
	sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE merge_file_path <> '' AND base_from_mysteel_chemical_classify_id IN(54,55,56,57) GROUP BY merge_file_path`
	_, err = o.Raw(sql).QueryRows(&items)
	return
}

// GetIndexRefreshMethanolByTimely 获取需要及时刷新的文件
func (m *BaseFromMysteelChemicalIndex) GetIndexRefreshMethanolByTimely() (items []*BaseFromMysteelChemicalIndex, err error) {
	o := orm.NewOrm()
	sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE merge_file_path <> '' AND is_refresh = 1 GROUP BY merge_file_path`
	_, err = o.Raw(sql).QueryRows(&items)
	return
}

type GetIndexByFrequencyReq struct {
	Frequency string
}

// GetIndexByFrequency 根据频率获取指标
func (m *BaseFromMysteelChemicalIndex) GetIndexByFrequency(frequency string) (items []*BaseFromMysteelChemicalIndex, err error) {
	o := orm.NewOrm()
	sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE frequency = ? AND merge_file_path = '' `
	_, err = o.Raw(sql, frequency).QueryRows(&items)
	return
}

type GetIndexRefreshWeekReq struct {
	NowWeekZn string
	StartTime string
	EndTime   string
}

func (m *BaseFromMysteelChemicalIndex) GetIndexRefreshWeek(nowWeekZn, startTime, endTime string) (items []*BaseFromMysteelChemicalIndex, err error) {
	where := `update_week = ? AND ((update_time >= ? AND update_time <= ?) OR (update_time2 >= ? AND update_time2 <= ?)) `
	o := orm.NewOrm()
	sql := fmt.Sprintf(`SELECT * FROM base_from_mysteel_chemical_index WHERE %s `, where)
	_, err = o.Raw(sql, nowWeekZn, startTime, endTime, startTime, endTime).QueryRows(&items)
	return
}

// UpdateIndexReq 根据编码更新指标请求体
type UpdateIndexReq struct {
	Item       *BaseFromMysteelChemicalIndex
	UpdateCols []string
}

func (m *BaseFromMysteelChemicalIndex) UpdateIndex(item *BaseFromMysteelChemicalIndex, updateCols []string) (err error) {
	if item == nil {
		return
	}
	if len(updateCols) == 0 {
		return
	}
	o := orm.NewOrm()
	_, err = o.Update(item, updateCols...)
	return
}

// MultiUpdateIndex 批量更新指标
func (m *BaseFromMysteelChemicalIndex) MultiUpdateIndex(items []UpdateIndexReq) (err error) {
	if len(items) == 0 {
		return
	}
	o := orm.NewOrm()
	for _, v := range items {
		if len(v.UpdateCols) == 0 {
			continue
		}
		if v.Item == nil {
			continue
		}
		if _, e := o.Update(v.Item, v.UpdateCols...); e != nil {
			err = fmt.Errorf("update err: %s", e.Error())
			return
		}
	}
	return
}

type HandleMysteelIndex struct {
	IndexName    string `description:"指标名称"`
	IndexCode    string `description:"指标编码"`
	Unit         string `description:"单位"`
	Source       string `description:"数据来源"`
	Frequency    string `description:"频度"`
	StartDate    string `description:"开始日期"`
	EndDate      string `description:"结束日期"`
	Describe     string `description:"指标描述"`
	UpdateDate   string `description:"更新日期"`
	ExcelDataMap map[string]string
}

type HandleMysteelIndexResp struct {
	List []*HandleMysteelIndex
}

// 钢联化工指标数据
type BaseFromMysteelChemicalData struct {
	BaseFromMysteelChemicalDataId  int64 `orm:"column(base_from_mysteel_chemical_data_id);pk"`
	BaseFromMysteelChemicalIndexId int64
	IndexCode                      string
	DataTime                       time.Time
	Value                          string
	UpdateDate                     string
	CreateTime                     time.Time `description:"创建时间"`
	ModifyTime                     time.Time `description:"修改时间"`
}

// GetIndexDataList 根据指标编码获取数据
func (d *BaseFromMysteelChemicalData) GetIndexDataList(indexCode string) (item []*BaseFromMysteelChemicalData, err error) {
	o := orm.NewOrm()
	sql := `SELECT * FROM base_from_mysteel_chemical_data WHERE index_code = ? `
	_, err = o.Raw(sql, indexCode).QueryRows(&item)
	return
}

// Update 修改
func (r *BaseFromMysteelChemicalData) Update(updateCols []string) (err error) {
	o := orm.NewOrm()
	_, err = o.Update(r, updateCols...)
	return
}

// Add 新增
func (r *BaseFromMysteelChemicalData) Add(list []BaseFromMysteelChemicalData) (err error) {
	o := orm.NewOrm()
	_, err = o.InsertMulti(len(list), list)
	return
}

type AddMysteelIndexResp struct {
	EdbCode                           string `description:"指标编码"`
	TerminalCode                      string `description:"指标终端编码"`
	BaseFromMysteelChemicalClassifyId int    `description:"指标分类"`
	SysUserId                         int    `description:"操作人id"`
	SysUserRealName                   string `description:"操作人真实名称"`
}

type GetIndexDetailReq struct {
	IndexCode string `description:"指标编码"`
}

// GetIndexCountByFrequency 获取未合并的指标总数量
func (m *BaseFromMysteelChemicalIndex) GetIndexCountByFrequency(condition string, pars []interface{}) (total int64, err error) {
	o := orm.NewOrm()

	sql := `SELECT count(1) AS total FROM base_from_mysteel_chemical_index WHERE 1=1  `
	if condition != "" {
		sql += condition
	}
	err = o.Raw(sql, pars).QueryRow(&total)

	return
}

// GetNoMergeIndexByFrequencyPage 获取未合并的指标列表
func (m *BaseFromMysteelChemicalIndex) GetIndexLimitByFrequency(frequency string, limit int) (items []*BaseFromMysteelChemicalIndex, err error) {
	o := orm.NewOrm()
	sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE frequency=? limit ? `
	_, err = o.Raw(sql, frequency, limit).QueryRows(&items)
	return
}

type GetIndexPageByFrequencyPageReq struct {
	Frequency    string
	EndDate      string
	StartSize    int
	PageSize     int
	TerminalCode string `description:"终端编码"`
}

// GetNoMergeIndexByFrequencyPage 获取未合并的指标列表
func (m *BaseFromMysteelChemicalIndex) GetIndexPageByFrequency(condition string, pars []interface{}, startSize, pageSize int) (items []*BaseFromMysteelChemicalIndex, err error) {
	o := orm.NewOrm()
	sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE 1=1 `
	if condition != "" {
		sql += condition
	}
	sql += ` LIMIT ?,?  `
	_, err = o.Raw(sql, pars, startSize, pageSize).QueryRows(&items)
	return
}

func (d *BaseFromMysteelChemicalData) GetMysteelIndexInfoMaxAndMinInfo(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_mysteel_chemical_data WHERE index_code=? `
	err = o.Raw(sql, indexCode).QueryRow(&item)

	if err != nil {
		return
	}

	// 获取最新值
	var latest_value float64
	sql = ` SELECT value AS latest_value FROM base_from_mysteel_chemical_data WHERE index_code=? ORDER BY data_time DESC LIMIT 1 `
	err = o.Raw(sql, indexCode).QueryRow(&latest_value)
	if err != nil {
		return
	}
	item.LatestValue = latest_value

	return
}

func (d *BaseFromMysteelChemicalData) ModifyMysteelIndexMaxAndMinInfo(indexCode string, item *EdbInfoMaxAndMinInfo) (err error) {
	o := orm.NewOrm()
	sql := ` UPDATE base_from_mysteel_chemical_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()
	return
}

// GetIndexByCondition 获取指标
func (m *BaseFromMysteelChemicalIndex) GetIndexByCondition(condition string, pars []interface{}) (items []*BaseFromMysteelChemicalIndex, err error) {
	o := orm.NewOrm()
	sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE 1=1 `
	if condition != "" {
		sql += condition
	}
	_, err = o.Raw(sql, pars).QueryRows(&items)
	return
}