123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702 |
- package eta_bridge
- import (
- "context"
- "encoding/json"
- "eta/eta_task/models/data_manage"
- "eta/eta_task/services/alarm_msg"
- "eta/eta_task/services/eta_hub"
- "eta/eta_task/utils"
- "fmt"
- "github.com/rdlucklib/rdluck_tools/uuid"
- "net/url"
- "strings"
- "sync"
- "time"
- )
- var lockSyncUser sync.Mutex
- func SyncUser(cont context.Context) (err error) {
- lockSyncUser.Lock()
- defer func() {
- if err != nil {
- tips := "SyncUser-定时将第三方的用户数据同步到ETA失败, ErrMsg:\n" + err.Error()
- utils.FileLog.Info(tips)
- go alarm_msg.SendAlarmMsg(tips, 3)
- }
- lockSyncUser.Unlock()
- }()
- uri := "/xy/user/pull"
- _, err, _ = HttpEtaBridgeGet(uri)
- if err != nil {
- return
- }
- return
- }
- type PushBaseParamReq struct {
- SerialID string `json:"serialID" description:"流水号"`
- TableCode string `json:"tableCode" description:"数据表编码"`
- Total int `json:"total" description:"本次落表数据总数"`
- IsEmailWarn int `json:"isEmailWarn" description:"是否发送预警邮件,(1-是 0-否)"`
- Data interface{} `json:"data" description:"报文体"`
- }
- type PushIndexParamDataReq struct {
- SourceIndexCode string `json:"source_index_code" description:"上游来源指标ID"`
- IndexCode string `json:"index_code" description:"数仓加工的指标ID,来源+来源ID,使用下划线连接,MST000ID00013242"`
- IndexName string `json:"index_name" description:"外部来源的指标名称"`
- IndexShortName string `json:"index_short_name" description:"系统内的指标名称/简称"`
- FrequenceName string `json:"frequence_name" description:"指标频度,如:日度、周度、月度"`
- UnitName string `json:"unit_name" description:"指标单位,如:元/吨、千克、立方米"`
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- AssetBeginDate string `json:"asset_begin_date" description:"业务字段,指标明细数据的业务日期开始时间;格式yyyy-mm-dd"`
- AssetEndDate string `json:"asset_end_date" description:"业务字段,指标明细数据的业务日期结束时间;格式yyyy-mm-dd"`
- CreateUser string `json:"create_user" description:"创建人姓名"`
- IndexCreateTime string `json:"index_create_time" description:"指标基础信息创建时间戳;格式yyyy-mm-dd hh:mi:ss"`
- UpdateUser string `json:"update_user" description:"更新人姓名"`
- DetailUpdateTime string `json:"detail_update_time" description:"指标明细信息更新时间戳;格式yyyy-mm-dd hh:mi:ss"`
- IndexUpdateTime string `json:"index_update_time" description:"指标基础信息更新时间戳;格式yyyy-mm-dd hh:mi:ss"`
-
-
- OrginSource string `json:"orgin_source" description:"外部数据原始来源,如国家统计局、钢联等"`
- OrginSysSource string `json:"orgin_sys_source" description:"外部来源系统,即数据供应商,如钢联、wind、同花顺"`
- SysSource string `json:"sys_source" description:"内部来源系统,如产研平台、市价平台"`
- SourceType string `json:"source_type" description:"数据接入方式,手工、接口、RPA"`
-
- Status int `json:"status" description:"逻辑删除:0-失效,1-有效"`
- }
- type PushIndexValueItemReq struct {
- Id string `json:"id"`
- IndexCode string `json:"index_code" description:"指标代码"`
- Value string `json:"value" description:"数值"`
- BusinessDate string `json:"business_date" description:"业务日期(数据日期)"`
- CreateTime string `json:"create_time" description:"数据进入ETA的时间"`
- UpdateTime string `json:"update_time" description:"eta库中修改数据的时间"`
- Status string `json:"status" description:"逻辑删除使用,0-禁用,1-启用"`
- }
- type PushClassifyItemReq struct {
- ClassifyId int `json:"classify_id" description:"自增id"`
- ClassifyType int `json:"classify_type" description:"分类类型,0:普通指标分类,1:预测指标分类"`
- ClassifyName string `json:"classify_name" description:"分类名称"`
- ParentId int `json:"parent_id" description:"父级id"`
- HasData int `json:"has_data" description:"是否存在指标数据,1:有,2:无"`
- CreateTime string `json:"create_time" description:"创建时间"`
- UpdateTime string `json:"update_time" description:"修改时间"`
- SysUserId int `json:"sys_user_id" description:"创建人id"`
- SysUserRealName string `json:"sys_user_real_name" description:"创建人姓名"`
- Level int `json:"level" description:"层级"`
- UniqueCode string `json:"unique_code" description:"唯一编码"`
- SortColumn int `json:"sort_column" description:"排序字段,越小越靠前,默认值:10"`
- }
- type PushEdbClassifyItemReq struct {
- Id string `json:"id" description:"唯一主键"`
- ClassifyId int `json:"classify_id" description:"目录分类ID"`
- IndexCode string `json:"index_code" description:"指标ID"`
- CreateTime string `json:"create_time" description:"创建时间"`
- CreateUser string `json:"create_user" description:"创建人"`
- UpdateTime string `json:"update_time" description:"修改时间"`
- UpdateUser string `json:"update_user" description:"修改人"`
- }
- var lockSyncClassify sync.Mutex
- func SyncClassifyList(cont context.Context) (err error) {
- lockSyncClassify.Lock()
- defer func() {
- if err != nil {
- tips := "SyncIndexList-定时同步ETA指标分类变更数据至第三方失败, ErrMsg:\n" + err.Error()
- utils.FileLog.Info(tips)
- go alarm_msg.SendAlarmMsg(tips, 3)
- }
- lockSyncClassify.Unlock()
- }()
- var condition string
- var pars []interface{}
-
- condition = " AND classify_type = ? "
- pars = append(pars, 0)
- list, err := data_manage.GetAllEdbClassifyListByCondition(condition, pars)
- if err != nil {
- utils.FileLog.Error("获取分类列表数据失败:" + err.Error())
- return
- }
- dataLimitList := make([][]PushClassifyItemReq, 0)
- dataList := make([]PushClassifyItemReq, 0)
- for _, v := range list {
- dataList = append(dataList, PushClassifyItemReq{
- ClassifyId: int(v.ClassifyID),
- ClassifyType: int(v.ClassifyType),
- ClassifyName: v.ClassifyName,
- ParentId: int(v.ParentID),
- HasData: int(v.HasData),
- CreateTime: v.CreateTime.Format(utils.FormatDateTime),
- UpdateTime: v.ModifyTime.Format(utils.FormatDateTime),
- SysUserId: int(v.SysUserID),
- SysUserRealName: v.SysUserRealName,
- Level: int(v.Level),
- UniqueCode: v.UniqueCode,
- SortColumn: int(v.Sort),
- })
- if len(dataList) >= 100 {
- dataLimitList = append(dataLimitList, dataList)
- dataList = make([]PushClassifyItemReq, 0)
- }
- }
- lenData := len(dataList)
- if lenData > 0 {
- dataLimitList = append(dataLimitList, dataList)
- }
- if len(dataLimitList) < 0 {
-
- return
- }
- errDataList := make([]PushClassifyItemReq, 0)
- errList := make([]string, 0)
- defer func() {
- if len(errList) > 0 {
- dataByte, err := json.Marshal(errDataList)
- if err != nil {
- dataByte = []byte("序列化分类列表数据失败" + err.Error())
- }
- utils.FileLog.Info("pushIndexClassify errList:%s;推送失败的分类列表数据列表:%s", strings.Join(errList, "\n"), string(dataByte))
- }
- }()
- for k, tmpDataList := range dataLimitList {
- req := PushBaseParamReq{
- SerialID: uuid.NewUUID().Hex32(),
- TableCode: "",
- Total: len(tmpDataList),
- IsEmailWarn: 0,
- Data: tmpDataList,
- }
- uri := "/xy/index/pushClassify"
- _, e, _ := HttpEtaBridgePost(uri, req)
- if e != nil {
- errList = append(errList, fmt.Sprintf("第%d组分类列表数据推送失败,postRefreshEdbData err: %s", k+1, e.Error()))
- errDataList = append(errDataList, dataList...)
- continue
- }
- }
- return
- }
- var lockSyncIndex sync.Mutex
- func SyncIndex(cont context.Context) (err error) {
- lockSyncIndex.Lock()
- defer func() {
- if err != nil {
- tips := "SyncIndex-定时同步ETA指标信息变更数据至第三方失败, ErrMsg:\n" + err.Error()
- utils.FileLog.Info(tips)
- go alarm_msg.SendAlarmMsg(tips, 3)
- }
- lockSyncIndex.Unlock()
- }()
-
- logMaxId, err := data_manage.GetEdbUpdateLogMaxId()
- if err != nil {
- return
- }
- var currLogId int64
-
- currLogId, err = utils.Rc.GetInt64(utils.CACHE_EDB_UPDATE_LOG_ID)
- err = fmt.Errorf(utils.RedisNoKeyErr)
- if err != nil {
-
- if err.Error() != utils.RedisNoKeyErr {
- return
- }
- err = nil
-
- currLogId, err = data_manage.GetEdbUpdateLogMaxHandleId()
- if err != nil {
- if err.Error() != utils.ErrNoRow() {
- utils.FileLog.Error("查找当前已经处理了的日志最大ID失败:" + err.Error())
- } else {
- err = nil
- }
- currLogId = 0
- }
- }
-
- for currId := currLogId; currId < logMaxId; {
- currId = handlePush(currId, logMaxId)
- }
- return
- }
- func handlePush(currLogIdStr, logMaxId int64) (lastId int64) {
- lastId = currLogIdStr
-
- var condition string
- var pars []interface{}
- condition += " AND id > ? AND id <= ?"
- pars = append(pars, currLogIdStr, logMaxId)
- list, err := data_manage.GetEdbUpdateLogByCondition(condition, pars)
- if err != nil {
- utils.FileLog.Error("获取变更日志失败:" + err.Error())
- return
- }
- pushIndexList := make([]*PushIndexParamDataReq, 0)
- pushEdbClassifyList := make([]*PushEdbClassifyItemReq, 0)
- pushIndexValueList := make([]*PushIndexValueItemReq, 0)
- idList := make([]int64, 0)
- for _, v := range list {
- lastId = v.Id
- idList = append(idList, v.Id)
- pushIndexData, pushEdbClassifyData, pushIndexValueData, tmpErr := handleData(v)
- if tmpErr != nil {
- err = tmpErr
- utils.FileLog.Error("获取待处理的数据失败:" + err.Error())
- continue
- }
- if pushIndexData != nil {
- pushIndexList = append(pushIndexList, pushIndexData)
- }
- if pushEdbClassifyData != nil {
- pushEdbClassifyList = append(pushEdbClassifyList, pushEdbClassifyData)
- }
- if pushIndexValueData != nil {
- pushIndexValueList = append(pushIndexValueList, pushIndexValueData)
- }
- }
- pushIndex(pushIndexList)
- pushIndexClassify(pushEdbClassifyList)
- pushIndexValue(pushIndexValueList)
-
- err = data_manage.HandleUpdateLogByIds(idList, time.Now().Format(utils.FormatDateTime))
- if err != nil {
- utils.FileLog.Error("批量处理指标更新记录失败:" + err.Error())
- }
- utils.Rc.Put(utils.CACHE_EDB_UPDATE_LOG_ID, lastId, 31*24*time.Hour)
- return
- }
- const pushBatchSize = 100
- func pushIndex(allPushList []*PushIndexParamDataReq) {
- lenDataList := len(allPushList)
- if lenDataList <= 0 {
- return
- }
- uri := utils.SyncIndexPath + "/pushIndexData"
- errDataList := make([]*PushIndexParamDataReq, 0)
- errList := make([]string, 0)
- defer func() {
- if len(errList) > 0 {
- dataByte, err := json.Marshal(errDataList)
- if err != nil {
- dataByte = []byte("序列化指标信息数据失败" + err.Error())
- }
- utils.FileLog.Info("pushIndex errList:%s;推送失败的指标信息数据列表:%s", strings.Join(errList, "\n"), string(dataByte))
- }
- }()
- dataLimitList := make([][]*PushIndexParamDataReq, 0)
- for i := 0; i < lenDataList; i += pushBatchSize {
- endIndex := min(i+pushBatchSize, lenDataList)
- tempSlice := allPushList[i:endIndex]
- dataLimitList = append(dataLimitList, tempSlice)
- }
- for k, dataList := range dataLimitList {
- req := PushBaseParamReq{
- SerialID: uuid.NewUUID().Hex32(),
- TableCode: "",
- Total: len(dataList),
- IsEmailWarn: 0,
- Data: dataList,
- }
- _, e, _ := HttpEtaBridgePost(uri, req)
- if e != nil {
- errList = append(errList, fmt.Sprintf("第%d组指标信息数据推送失败,postRefreshEdbData err: %s", k+1, e.Error()))
- errDataList = append(errDataList, dataList...)
- continue
- }
- }
- }
- func pushIndexValue(allPushList []*PushIndexValueItemReq) {
- lenDataList := len(allPushList)
- if lenDataList <= 0 {
- return
- }
- uri := utils.SyncIndexPath + "/pushIndexValue"
- errDataList := make([]*PushIndexValueItemReq, 0)
- errList := make([]string, 0)
- defer func() {
- if len(errList) > 0 {
- dataByte, err := json.Marshal(errDataList)
- if err != nil {
- dataByte = []byte("序列化指标明细数据失败" + err.Error())
- }
- utils.FileLog.Info("pushIndexValue errList:%s;推送失败的指标信息数据列表:%s", strings.Join(errList, "\n"), string(dataByte))
- }
- }()
- dataLimitList := make([][]*PushIndexValueItemReq, 0)
- for i := 0; i < lenDataList; i += pushBatchSize {
- endIndex := min(i+pushBatchSize, lenDataList)
- tempSlice := allPushList[i:endIndex]
- dataLimitList = append(dataLimitList, tempSlice)
- }
- for k, dataList := range dataLimitList {
- req := PushBaseParamReq{
- SerialID: uuid.NewUUID().Hex32(),
- TableCode: "",
- Total: len(dataList),
- IsEmailWarn: 0,
- Data: dataList,
- }
- _, e, _ := HttpEtaBridgePost(uri, req)
- if e != nil {
- errList = append(errList, fmt.Sprintf("第%d组指标明细数据推送失败,postRefreshEdbData err: %s", k+1, e.Error()))
- errDataList = append(errDataList, dataList...)
- continue
- }
- }
- return
- }
- func pushIndexClassify(allPushList []*PushEdbClassifyItemReq) {
- lenDataList := len(allPushList)
- if lenDataList <= 0 {
- return
- }
- uri := utils.SyncIndexPath + "/pushEdbClassify"
- errDataList := make([]*PushEdbClassifyItemReq, 0)
- errList := make([]string, 0)
- defer func() {
- if len(errList) > 0 {
- dataByte, err := json.Marshal(errDataList)
- if err != nil {
- dataByte = []byte("序列化指标明细数据失败" + err.Error())
- }
- utils.FileLog.Info("pushIndexClassify errList:%s;推送失败的指标所属分类明细数据列表:%s", strings.Join(errList, "\n"), string(dataByte))
- }
- }()
- dataLimitList := make([][]*PushEdbClassifyItemReq, 0)
- for i := 0; i < lenDataList; i += pushBatchSize {
- endIndex := min(i+pushBatchSize, lenDataList)
- tempSlice := allPushList[i:endIndex]
- dataLimitList = append(dataLimitList, tempSlice)
- }
- for k, dataList := range dataLimitList {
- req := PushBaseParamReq{
- SerialID: uuid.NewUUID().Hex32(),
- TableCode: "",
- Total: len(dataList),
- IsEmailWarn: 0,
- Data: dataList,
- }
- _, e, _ := HttpEtaBridgePost(uri, req)
- if e != nil {
- errList = append(errList, fmt.Sprintf("第%d组指标所属分类数据推送失败,postRefreshEdbData err: %s", k+1, e.Error()))
- errDataList = append(errDataList, dataList...)
- continue
- }
- }
- return
- }
- func min(a, b int) int {
- if a < b {
- return a
- }
- return b
- }
- var lockGetCrmIndex sync.Mutex
- func SyncXyCrmIndex(cont context.Context) (err error) {
- lockGetCrmIndex.Lock()
- errMsgList := make([]string, 0)
- defer func() {
- if err != nil {
- tips := "SyncXyCrmIndex-定时同步CRM数据到ETA平台失败, ErrMsg:\n" + err.Error()
- utils.FileLog.Info(tips)
- go alarm_msg.SendAlarmMsg(tips, 3)
- }
- if len(errMsgList) > 0 {
- tips := "SyncXyCrmIndex-定时同步CRM数据到ETA平台失败, ErrMsg:\n" + strings.Join(errMsgList, "\n")
- utils.FileLog.Info(tips)
- go alarm_msg.SendAlarmMsg(tips, 3)
- }
- lockGetCrmIndex.Unlock()
- fmt.Println("end SyncXyCrmIndex")
- }()
-
- if utils.SyncCrmAssetPkgCd == `` || utils.SyncCrmDataSourceType == `` {
- return
- }
- var lastUpdateTimeStr string
- nowTimeStr := time.Now().Format(utils.FormatDateTimeUnSpaceV2)
- key := data_manage.CrmIndexLastUpdateTime
- sysInteractionLog, err := data_manage.GetBusinessSysInteractionLogByKey(key)
- if err != nil {
- if err.Error() != utils.ErrNoRow() {
- return
- }
-
- } else {
- if sysInteractionLog.InteractionVal != `` {
- lastUpdateTimeStr = sysInteractionLog.InteractionVal
- }
- }
- syncCrmAssetPkgCd := utils.SyncCrmAssetPkgCd
- syncCrmAssetPkgCdList := strings.Split(syncCrmAssetPkgCd, ",")
- for _, assetPkgCd := range syncCrmAssetPkgCdList {
- err, errMsgList = syncCrmIndex(assetPkgCd, 1, utils.SyncCrmIndexNum, lastUpdateTimeStr)
- }
-
- modifyCrmIndexLastUpdateTime(nowTimeStr)
- return
- }
- func syncCrmIndex(assetPkgCd string, currIndex, pageSize int, baseLastUpdateTimeStr string) (err error, errMsgList []string) {
- errMsgList = make([]string, 0)
- lastUpdateTimeStr := baseLastUpdateTimeStr
- if lastUpdateTimeStr != `` {
- lastUpdateTimeStr = url.QueryEscape(lastUpdateTimeStr)
- }
- uri := fmt.Sprintf("%s/getCrmData?index_pkg_code=%s&data_source_type=%s¤t_index=%d&page_size=%d&detail_last_update_start_time=%s", utils.SyncCrmIndexPath, assetPkgCd, utils.SyncCrmDataSourceType, currIndex, pageSize, lastUpdateTimeStr)
- bResult, err, _ := HttpEtaBridgeGet(uri)
- if err != nil {
- return
- }
- result := new(EtaBridgeDataRespAndBusinessData)
- err = json.Unmarshal(bResult, &result)
- if err != nil {
- err = fmt.Errorf("result unmarshal err: %s\nresult: %s", err.Error(), string(bResult))
- utils.FileLog.Info("桥接服务get请求失败:\n" + string(bResult))
- return
- }
-
- for _, v := range result.Data.List {
- tmpErr := pushCrmDataToHub(v)
- if tmpErr != nil {
- errMsgList = append(errMsgList, tmpErr.Error())
- }
- }
-
- if currIndex < result.Data.Paging.Pages {
- _, tmpErrMsgList := syncCrmIndex(assetPkgCd, currIndex+1, utils.SyncCrmIndexNum, baseLastUpdateTimeStr)
- errMsgList = append(errMsgList, tmpErrMsgList...)
- }
- return
- }
- func pushCrmDataToHub(data interface{}) (err error) {
- uri := `/edb/push`
- _, err, _ = eta_hub.HttpEtaHubPost(uri, data)
-
-
-
-
-
-
-
- return
- }
- func modifyCrmIndexLastUpdateTime(lastUpdateTime string) {
- var err error
- defer func() {
- if err != nil {
- utils.FileLog.Error("修改binlog文件名称和位置异常,lastUpdateTime", lastUpdateTime, ",err:", err)
- }
- }()
-
- key := data_manage.CrmIndexLastUpdateTime
- fileNameLog, err := data_manage.GetBusinessSysInteractionLogByKey(key)
- if err != nil {
- if err.Error() != utils.ErrNoRow() {
- return
- }
- err = nil
- fileNameLog = &data_manage.BusinessSysInteractionLog{
-
- InteractionKey: key,
- InteractionVal: lastUpdateTime,
- Remark: "crm拉取数据的最近更新时间",
- ModifyTime: time.Now(),
- CreateTime: time.Now(),
- }
- err = fileNameLog.Create()
- if err != nil {
- return
- }
- } else {
- fileNameLog.InteractionVal = lastUpdateTime
- fileNameLog.ModifyTime = time.Now()
- err = fileNameLog.Update([]string{"InteractionVal", "ModifyTime"})
- if err != nil {
- return
- }
- }
- return
- }
|