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:"终端编码"` } type MysteelChemicalAPiCheck struct { IsEnable bool ErrMsg string } // 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) GetBatchIndexItem(indexCodes []string) (items []*BaseFromMysteelChemicalIndex, err error) { if len(indexCodes) <= 0 { return } o := orm.NewOrm() sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE index_code IN (%s) ` holder := make([]string, 0, len(indexCodes)) for range indexCodes { holder = append(holder, "?") } sql = fmt.Sprintf(sql, strings.Join(holder, ",")) _, err = o.Raw(sql, indexCodes).QueryRows(&items) 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 } // AddV2 新增 func (r *BaseFromMysteelChemicalData) AddV2(list []*BaseFromMysteelChemicalData) (err error) { if len(list) == 0 { return } o := orm.NewOrm() _, err = o.InsertMulti(500, 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 } type BaseFromMysteelChemicalRecord struct { BaseFromMysteelChemicalRecordId int64 `orm:"column(base_from_mysteel_chemical_record_id);pk"` BaseFromMysteelChemicalIndexId int64 OldIndexName string `description:"原始名称"` NewIndexName string `description:"新的名称"` CreateTime time.Time `description:"记录创建时间"` Timestamp int64 `description:"记录创建时间戳"` } func (m *BaseFromMysteelChemicalRecord) AddBaseFromMysteelChemicalRecord() (err error) { o := orm.NewOrm() _, err = o.Insert(m) return } type MySteelChemicalApiResp struct { Code string `json:"code" description:"200成功,其他失败"` Success bool `json:"success" description:"true 成功,false 失败"` Timestamp int64 `json:"timestamp" description:"时间戳"` Message string `json:"message" description:"显示执行信息"` Data []*MySteelChemicalApiData `json:"data" description:"数据"` } type MySteelChemicalApiInfoResp struct { Code string `json:"code" description:"200成功,其他失败"` Success bool `json:"success" description:"true 成功,false 失败"` Timestamp int64 `json:"timestamp" description:"时间戳"` Message string `json:"message" description:"显示执行信息"` Data *MySteelChemicalApiInfo `json:"data" description:"数据"` } type MySteelChemicalApiInfo struct { Total int `json:"total" description:"总条数"` Pages int `json:"pages" description:"总页数"` List []*MySteelChemicalApiInfoItem `json:"list" description:"数据列表"` } type MySteelChemicalApiInfoItem struct { IndexCode string `json:"INDEX_CODE"` IndexName string `json:"INDEX_NAME"` FrequencyName string `json:"FREQUENCY_NAME"` UnitName string `json:"UNIT_NAME"` } type MySteelChemicalApiData struct { IndexCode string `json:"INDEX_CODE"` DataList []*MySteelChemicalApiDataList `json:"dataList"` } type MySteelChemicalApiDataList struct { PublishTime int64 `json:"PUBLISH_TIME"` IndexCode string `json:"INDEX_CODE"` DataDate string `json:"DATA_DATE"` DataValue float64 `json:"DATA_VALUE"` }