123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712 |
- package services
- import (
- "errors"
- "eta/eta_index_lib/logic"
- "eta/eta_index_lib/models"
- "eta/eta_index_lib/models/mgo"
- "eta/eta_index_lib/services/alarm_msg"
- "eta/eta_index_lib/utils"
- "fmt"
- "github.com/qiniu/qmgo"
- "go.mongodb.org/mongo-driver/bson"
- "strings"
- "time"
- )
- func HandleBusinessIndex(indexReq *models.AddBusinessIndexReq) (resp models.BaseFromBusinessIndexResp, err error) {
- defer func() {
- if err != nil {
-
- dataUpdateResult := 2
- dataUpdateFailedReason := "服务异常"
- edbInfo, e := models.GetEdbInfoByEdbCode(utils.DATA_SOURCE_MYSTEEL_CHEMICAL, indexReq.IndexCode)
- if e == nil {
-
- _ = AddEdbInfoUpdateLog(edbInfo.EdbInfoId, 2, err.Error(), dataUpdateResult, dataUpdateFailedReason, 1, 0)
- }
- }
- }()
-
- if indexReq.DataList == nil || len(indexReq.DataList) <= 0 {
- return
- }
-
- if !strings.Contains(indexReq.Frequency, "度") {
- indexReq.Frequency = indexReq.Frequency + "度"
- }
- if !utils.VerifyFrequency(indexReq.Frequency) {
- err = errors.New("指标频度不合法:" + indexReq.Frequency)
- return
- }
-
- sourceObj := new(models.EdbBusinessSource)
- sourceItem, err := sourceObj.GetEdbBusinessSourceItem(indexReq.SourceName)
- if err != nil {
- if err.Error() == utils.ErrNoRow() {
- sourceItem = &models.EdbBusinessSource{
- EdbBusinessSourceId: 0,
- SourceName: indexReq.SourceName,
- CreateTime: time.Now(),
- }
- err = sourceItem.Add()
- } else {
- return
- }
- }
-
- indexObj := new(models.BaseFromBusinessIndex)
- if indexReq.IndexCode == `` {
-
- currId, tmpErr := indexObj.GetMaxId()
- if tmpErr != nil {
- err = tmpErr
- return
- }
- indexReq.IndexCode = fmt.Sprintf("SELF%07d", currId+1)
- }
-
- item, err := indexObj.GetIndexItem(indexReq.IndexCode)
- if err != nil {
- if err.Error() != utils.ErrNoRow() {
- return
- }
-
- item = &models.BaseFromBusinessIndex{
- BaseFromBusinessIndexId: 0,
- IndexCode: indexReq.IndexCode,
- IndexName: indexReq.IndexName,
- Unit: indexReq.Unit,
- Frequency: indexReq.Frequency,
- Source: int(sourceItem.EdbBusinessSourceId),
- SourceName: sourceItem.SourceName,
-
-
- Remark: indexReq.Remark,
- BaseModifyTime: time.Now(),
- DataUpdateTime: time.Now(),
- CreateTime: time.Now(),
- ModifyTime: time.Now(),
- }
- err = item.Add()
- if err != nil {
- fmt.Println("add err:" + err.Error())
- return
- }
- } else {
- updateCols := make([]string, 0)
- if item.IndexName != indexReq.IndexName {
- item.IndexName = indexReq.IndexName
- updateCols = append(updateCols, "IndexName")
- }
- if item.Unit != indexReq.Unit {
- item.Unit = indexReq.Unit
- updateCols = append(updateCols, "Unit")
- }
- if item.Frequency != indexReq.Frequency {
- item.Frequency = indexReq.Frequency
- updateCols = append(updateCols, "Frequency")
- }
- if item.Source != int(sourceItem.EdbBusinessSourceId) {
- item.Source = int(sourceItem.EdbBusinessSourceId)
- item.SourceName = sourceItem.SourceName
- updateCols = append(updateCols, "Source", "SourceName")
- }
- if len(updateCols) > 0 {
- item.BaseModifyTime = time.Now()
- item.ModifyTime = time.Now()
- updateCols = append(updateCols, "BaseModifyTime", "ModifyTime")
- err = item.Update(updateCols)
- if err != nil {
- fmt.Println("update index err:" + err.Error())
- return
- }
- }
- }
-
-
- var reqMinDate time.Time
- if utils.UseMongo {
- reqMinDate, err = handleBusinessDataByMongo(item, indexReq.DataList)
- } else {
- reqMinDate, err = handleBusinessDataByMysql(item, indexReq.DataList)
- }
- if err != nil {
- return
- }
-
- go refreshEdbBusiness(item.IndexCode, reqMinDate)
- resp = models.BaseFromBusinessIndexResp{
- IndexCode: item.IndexCode,
- IndexName: item.IndexName,
- Unit: item.Unit,
- Frequency: item.Frequency,
- SourceName: item.SourceName,
- }
- return
- }
- func handleBusinessDataByMongo(item *models.BaseFromBusinessIndex, reqDataList []models.AddBusinessDataReq) (reqMinDate time.Time, err error) {
- mogDataObj := new(mgo.BaseFromBusinessData)
-
- exitDataList, err := mogDataObj.GetAllDataList(bson.M{"index_code": item.IndexCode}, []string{"data_time"})
- if err != nil {
- fmt.Println("GetIndexDataList Err:" + err.Error())
- return
- }
-
- exitDataMap := make(map[string]*mgo.BaseFromBusinessData)
- for _, v := range exitDataList {
- exitDataMap[v.DataTime.Format(utils.FormatDate)] = v
- }
-
- addDataList := make([]interface{}, 0)
- updateDataList := make([]mgo.BaseFromBusinessData, 0)
-
-
- for _, data := range reqDataList {
- dateTime, tmpErr := utils.DealExcelDate(data.Date)
- if tmpErr != nil {
- fmt.Println("time.ParseInLocation Err:" + tmpErr.Error())
- err = tmpErr
- return
- }
-
- if reqMinDate.IsZero() || reqMinDate.After(dateTime) {
- reqMinDate = dateTime
- }
- date := dateTime.Format(utils.FormatDate)
- findData, ok := exitDataMap[date]
- if !ok {
- addDataList = append(addDataList, mgo.BaseFromBusinessData{
- BaseFromBusinessIndexId: item.BaseFromBusinessIndexId,
- IndexCode: item.IndexCode,
- DataTime: dateTime,
- Value: data.Value,
- CreateTime: time.Now(),
- ModifyTime: time.Now(),
-
- })
- continue
- }
-
- if findData.Value != data.Value {
- findData.Value = data.Value
- updateDataList = append(updateDataList, *findData)
- }
- }
-
- var isIndexUpdateOrAdd bool
-
- {
- coll := mogDataObj.GetCollection()
- if len(addDataList) > 0 {
- isIndexUpdateOrAdd = true
- err = mogDataObj.BatchInsertDataByColl(coll, 500, addDataList)
- if err != nil {
- fmt.Println("mogDataObj.HandleData() Err:" + err.Error())
- return
- }
- }
- if len(updateDataList) > 0 {
- isIndexUpdateOrAdd = true
- for _, v := range updateDataList {
- err = mogDataObj.UpdateDataByColl(coll, bson.M{"_id": v.ID}, bson.M{"$set": bson.M{"value": v.Value, "modify_time": v.ModifyTime}})
- if err != nil {
- fmt.Println("UpdateDataByColl:Err:" + err.Error())
- return
- }
- }
- }
- }
-
-
-
-
-
-
-
-
- indexMaxAndMinInfo, err := item.GetEdbInfoMaxAndMinInfo(item.IndexCode)
- if err != nil {
- return
- }
- if err == nil && indexMaxAndMinInfo != nil {
- e := item.ModifyIndexMaxAndMinInfo(item.IndexCode, indexMaxAndMinInfo, isIndexUpdateOrAdd)
- if e != nil {
- fmt.Println("ModifyIndexMaxAndMinInfo Err:" + e.Error())
- }
- }
- return
- }
- func handleBusinessDataByMysql(item *models.BaseFromBusinessIndex, reqDataList []models.AddBusinessDataReq) (reqMinDate time.Time, err error) {
- businessDataObj := new(models.BaseFromBusinessData)
- var condition []string
- var pars []interface{}
- condition = append(condition, "index_code = ?")
- pars = append(pars, item.IndexCode)
-
- exitDataList, err := businessDataObj.GetAllDataList(condition, pars, "data_time ASC")
- if err != nil {
- fmt.Println("GetIndexDataList Err:" + err.Error())
- return
- }
-
- exitDataMap := make(map[string]*models.BaseFromBusinessData)
- for _, v := range exitDataList {
- exitDataMap[v.DataTime.Format(utils.FormatDate)] = v
- }
-
- addDataList := make([]*models.BaseFromBusinessData, 0)
- updateDataList := make([]*models.BaseFromBusinessData, 0)
-
-
- for _, data := range reqDataList {
- dateTime, tmpErr := utils.DealExcelDate(data.Date)
- if tmpErr != nil {
- fmt.Println("time.ParseInLocation Err:" + tmpErr.Error())
- err = tmpErr
- return
- }
-
- if reqMinDate.IsZero() || reqMinDate.After(dateTime) {
- reqMinDate = dateTime
- }
- date := dateTime.Format(utils.FormatDate)
- findData, ok := exitDataMap[date]
- if !ok {
- addDataList = append(addDataList, &models.BaseFromBusinessData{
- BaseFromBusinessIndexId: int(item.BaseFromBusinessIndexId),
- IndexCode: item.IndexCode,
- DataTime: dateTime,
- Value: data.Value,
- CreateTime: time.Now(),
- ModifyTime: time.Now(),
-
- })
- continue
- }
-
- if findData.Value != data.Value {
- findData.Value = data.Value
- findData.ModifyTime = time.Now()
- updateDataList = append(updateDataList, findData)
- }
- }
-
- var isIndexUpdateOrAdd bool
-
- {
- if len(addDataList) > 0 {
- isIndexUpdateOrAdd = true
- }
- if len(updateDataList) > 0 {
- isIndexUpdateOrAdd = true
- }
- err = businessDataObj.HandleData(addDataList, updateDataList)
- if err != nil {
- fmt.Println("UpdateDataByColl:Err:" + err.Error())
- return
- }
- }
-
- indexMaxAndMinInfo, err := item.GetEdbInfoMaxAndMinInfo(item.IndexCode)
- if err != nil {
- return
- }
- if err == nil && indexMaxAndMinInfo != nil {
- e := item.ModifyIndexMaxAndMinInfo(item.IndexCode, indexMaxAndMinInfo, isIndexUpdateOrAdd)
- if e != nil {
- fmt.Println("ModifyIndexMaxAndMinInfo Err:" + e.Error())
- }
- }
- return
- }
- type DelBusinessIndexResp struct {
- IsDeleteEdbCodeList []string `description:"已经删除了的指标编码"`
- NoDeleteEdbCodeList []string `description:"未删除的指标编码"`
- }
- func DelBusinessIndex(indexCodeList []string) (joinEdbCodeList, needDelEdbCodeList []string, err error, errMsg string) {
- defer func() {
- if err != nil {
- fmt.Println("DelBusinessIndex Err:" + err.Error())
- }
- }()
- errMsg = "删除失败"
- if len(indexCodeList) < 0 {
- errMsg = "指标编码不允许为空"
- err = errors.New(errMsg)
- return
- }
-
- indexObj := new(models.BaseFromBusinessIndex)
-
- baseIndexList, err := indexObj.GetIndexItemList(indexCodeList)
- if err != nil {
- return
- }
-
- joinEdbCodeList = make([]string, 0)
- joinEdbMap := make(map[string]string)
-
- needDelEdbCodeList = make([]string, 0)
-
- tmpEdbInfoList, err := models.GetEdbInfoByEdbCodeList(utils.DATA_SOURCE_BUSINESS, indexCodeList)
- if err != nil {
- return
- }
- for _, v := range tmpEdbInfoList {
- joinEdbCodeList = append(joinEdbCodeList, v.EdbCode)
- joinEdbMap[v.EdbCode] = v.EdbCode
- }
- for _, v := range baseIndexList {
- _, ok := joinEdbMap[v.IndexCode]
- if !ok {
- needDelEdbCodeList = append(needDelEdbCodeList, v.IndexCode)
- }
- }
-
- if len(needDelEdbCodeList) <= 0 {
- return
- }
-
- err = indexObj.DelIndexItemList(needDelEdbCodeList)
- if err != nil {
- fmt.Println("删除自有指标失败, Err:" + err.Error())
- return
- }
- if utils.UseMongo {
-
- mogDataObj := new(mgo.BaseFromBusinessData)
- err = mogDataObj.RemoveMany(bson.M{"index_code": bson.M{"$in": needDelEdbCodeList}})
- } else {
- var condition []string
- var pars []interface{}
- delNum := len(needDelEdbCodeList)
- if delNum > 0 {
- condition = append(condition, "index_code in ("+utils.GetOrmInReplace(delNum)+")")
- pars = append(pars, needDelEdbCodeList)
- businessDataObj := models.BaseFromBusinessData{}
- err = businessDataObj.DelDataByCond(condition, pars)
- }
- }
- if err != nil {
- fmt.Println("删除自有指标明细数据 Err:" + err.Error())
- return
- }
- return
- }
- func DelBusinessIndexData(indexCode string, startDate, endDate string) (err error, errMsg string) {
- defer func() {
- if err != nil {
- fmt.Println("DelBusinessIndexData Err:" + err.Error())
- }
- }()
- errMsg = "删除失败"
- if indexCode == `` {
- errMsg = "指标编码不允许为空"
- err = errors.New(errMsg)
- return
- }
- if startDate == `` && endDate == `` {
- errMsg = "开始日期和结束日期不允许同时为空"
- err = errors.New(errMsg)
- return
- }
-
- indexObj := new(models.BaseFromBusinessIndex)
-
- item, err := indexObj.GetIndexItem(indexCode)
- if err != nil {
- if err.Error() != utils.ErrNoRow() {
- return
- }
- err = nil
- return
- }
-
- var reqMinDate time.Time
- var startDateTime, endDateTime time.Time
- if startDate != `` {
-
- startDateTime, err = time.ParseInLocation(utils.FormatDate, startDate, time.Local)
- if err != nil {
- return
- }
-
- if reqMinDate.IsZero() || reqMinDate.After(startDateTime) {
- reqMinDate = startDateTime
- }
- }
- if endDate != `` {
-
- endDateTime, err = time.ParseInLocation(utils.FormatDate, endDate, time.Local)
- if err != nil {
- return
- }
-
- if reqMinDate.IsZero() || reqMinDate.After(endDateTime) {
- reqMinDate = endDateTime
- }
- }
-
- var indexMaxAndMinInfo *models.EdbInfoMaxAndMinInfo
- if utils.UseMongo {
- indexMaxAndMinInfo, err = delBusinessIndexDataByMongo(item, startDateTime, endDateTime)
- } else {
- indexMaxAndMinInfo, err = delBusinessIndexDataByMysql(item, startDateTime, endDateTime)
- }
- if err != nil {
- return
- }
-
- if indexMaxAndMinInfo != nil {
- e := item.ModifyIndexMaxAndMinInfo(item.IndexCode, indexMaxAndMinInfo, true)
- if e != nil {
- fmt.Println("ModifyIndexMaxAndMinInfo Err:" + e.Error())
- }
- }
-
- go refreshEdbBusiness(item.IndexCode, reqMinDate)
- return
- }
- func delBusinessIndexDataByMongo(item *models.BaseFromBusinessIndex, startDateTime, endDateTime time.Time) (indexMaxAndMinInfo *models.EdbInfoMaxAndMinInfo, err error) {
- defer func() {
- if err != nil {
- utils.FileLog.Error("delBusinessIndexDataByMongo 删除自有指标明细数据 Err:" + err.Error())
- }
- }()
-
- queryConditions := bson.M{
- "index_code": item.IndexCode,
- }
- dateCondition, err := mgo.BuildDateTimeCondition(startDateTime, endDateTime)
- if err != nil {
- return
- }
- if len(dateCondition) > 0 {
- queryConditions["data_time"] = dateCondition
- }
-
- mogDataObj := new(mgo.BaseFromBusinessData)
- err = mogDataObj.RemoveMany(queryConditions)
- if err != nil {
- return
- }
-
- indexMaxAndMinInfo, err = item.GetEdbInfoMaxAndMinInfo(item.IndexCode)
-
- if err != nil && !errors.Is(err, qmgo.ErrNoSuchDocuments) {
- return
- }
-
- err = nil
- return
- }
- func delBusinessIndexDataByMysql(item *models.BaseFromBusinessIndex, startDateTime, endDateTime time.Time) (indexMaxAndMinInfo *models.EdbInfoMaxAndMinInfo, err error) {
- defer func() {
- if err != nil {
- utils.FileLog.Error("delBusinessIndexDataByMysql 删除自有指标明细数据 Err:" + err.Error())
- }
- }()
-
- var condition []string
- var pars []interface{}
- condition = append(condition, "index_code = ? ")
- pars = append(pars, item.IndexCode)
- if !startDateTime.IsZero() {
- condition = append(condition, " data_time >= ? ")
- pars = append(pars, startDateTime.Format(utils.FormatDate))
- }
- if !endDateTime.IsZero() {
- condition = append(condition, " data_time <= ? ")
- pars = append(pars, endDateTime.Format(utils.FormatDate))
- }
-
- businessDataObj := new(models.BaseFromBusinessData)
- err = businessDataObj.DelDataByCond(condition, pars)
- if err != nil {
- return
- }
-
- indexMaxAndMinInfo, err = item.GetEdbInfoMaxAndMinInfo(item.IndexCode)
-
- if err != nil && err.Error() != utils.ErrNoRow() {
- return
- }
-
- err = nil
- return
- }
- func refreshEdbBusiness(indexCode string, reqMinDate time.Time) {
- var indexErr error
- var errMsg string
- defer func() {
- if indexErr != nil {
- tips := fmt.Sprintf("自有数据刷新-ETA指标刷新异常, 指标编码: %s, err: %s, 错误信息:%s", indexCode, indexErr.Error(), errMsg)
- alarm_msg.SendAlarmMsg(tips, 3)
- }
- }()
- edbInfo, e := models.GetEdbInfoByEdbCode(utils.DATA_SOURCE_BUSINESS, indexCode)
- if e != nil && e.Error() != utils.ErrNoRow() {
- indexErr = e
- return
- }
- if edbInfo != nil {
- startDate := ``
- if reqMinDate.IsZero() {
- startDate = edbInfo.EndDate
- } else {
- startDate = reqMinDate.Format(utils.FormatDate)
- }
- params := models.RefreshBaseParams{
- EdbInfo: edbInfo,
- StartDate: startDate,
- }
- obj := models.Business{}
- indexErr, errMsg = obj.Refresh(params)
- if indexErr != nil {
- return
- }
-
- indexErr = obj.UnifiedModifyEdbInfoMaxAndMinInfo(edbInfo)
- if indexErr != nil {
- return
- }
-
- go logic.UpdateEs(edbInfo.EdbInfoId)
- }
- }
|