eta_task/services/edb_relation.go

package services

import (
	"encoding/json"
	"eta/eta_task/models/data_manage"
	"eta/eta_task/models/data_manage/cross_variety"
	"eta/eta_task/models/fe_calendar"
	"eta/eta_task/models/sandbox"
	"eta/eta_task/services/alarm_msg"
	"eta/eta_task/utils"
	"fmt"
	"time"
)

func InitEdbRelation() {
	//InitChartEdbRelation()
	//InitChartCrossVariety()
	//initCalendarIndicatorRelation()
	initSandBoxEdbRelation()
}
func InitChartEdbRelation() {
	fmt.Println("开始处理图表中的指标引用")
	var err error
	var addNum int
	defer func() {
		if err != nil {
			msg := fmt.Sprintf("初始化指标在图表中的引用失败 InitChartEdbRelation  err: %v", err)
			utils.FileLog.Info(msg)
			fmt.Println(msg)
			go alarm_msg.SendAlarmMsg(msg, 3)
		}
	}()
	//查询chart_edb_mapping 表
	total, err := data_manage.GetChartEdbMappingTotal()
	if err != nil {
		err = fmt.Errorf("查询图表关联指标失败 err: %v", err)
		return
	}
	if total == 0 {
		return
	}
	//分页查询，每次处理500条记录
	pageSize := 500
	totalPage := (total + pageSize - 1) / pageSize // 使用整数除法，并添加一页以防有余数
	addList := make([]*data_manage.EdbInfoRelation, 0)
	//查询图表列表
	for i := 0; i < totalPage; i += 1 {
		startSize := i * pageSize
		list, e := data_manage.GetChartEdbMappingList(startSize, pageSize)
		if e != nil {
			err = fmt.Errorf("查询图表关联指标列表失败 Err:%s", e)
			return
		}
		if len(list) == 0 {
			break
		}

		edbInfoIds := make([]int, 0)
		for _, v := range list {
			edbInfoIds = append(edbInfoIds, v.EdbInfoId)
		}
		// 查询指标信息表
		edbInfoList, e := data_manage.GetEdbInfoByIdListAndSource(edbInfoIds, []int{utils.DATA_SOURCE_WIND, utils.DATA_SOURCE_MYSTEEL_CHEMICAL})
		if e != nil {
			err = fmt.Errorf("查询指标信息列表失败 Err:%s", e)
			return
		}
		if len(edbInfoList) == 0 {
			continue
		}
		edbInfoMap := make(map[int]*data_manage.EdbInfo)
		for _, v := range edbInfoList {
			edbInfoMap[v.EdbInfoId] = v
		}

		// 筛选有用的图表
		finalList := make([]*data_manage.ChartEdbMapping, 0)
		chartIds := make([]int, 0)
		for _, v := range list {
			if _, ok2 := edbInfoMap[v.EdbInfoId]; !ok2 {
				continue
			}
			finalList = append(finalList, v)
			chartIds = append(chartIds, v.ChartInfoId)
		}
		if len(chartIds) == 0 {
			continue
		}
		// 查询图表信息
		chartInfoList, e := data_manage.GetChartInfoByChartInfoIds(chartIds)
		if e != nil {
			err = fmt.Errorf("查询图表信息列表失败 Err:%s", e)
			return
		}
		chartInfoMap := make(map[int]*data_manage.ChartInfo)
		for _, v := range chartInfoList {
			chartInfoMap[v.ChartInfoId] = v
		}

		//查询引用关系列表，
		chartEdbRelationList, e := data_manage.GetEdbInfoRelationByReferObjectIds(chartIds, utils.EDB_RELATION_CHART)
		if e != nil {
			err = fmt.Errorf("查询图表引用关系列表失败 Err:%s", e)
			return
		}
		existRelationMap := make(map[string]struct{})
		for _, v := range chartEdbRelationList {
			name := fmt.Sprintf("%d-%d", v.ReferObjectId, v.EdbInfoId)
			existRelationMap[name] = struct{}{}
		}
		for _, v := range finalList {
			nowTime := time.Now()
			name := fmt.Sprintf("%d-%d", v.ChartInfoId, v.EdbInfoId)
			if _, ok := existRelationMap[name]; !ok {
				//查询图表信息
				chartInfo, ok1 := chartInfoMap[v.ChartInfoId]
				if !ok1 {
					continue
				}
				edbInfo, ok2 := edbInfoMap[v.EdbInfoId]
				if !ok2 {
					continue
				}
				//todo 引用时间
				tmp := &data_manage.EdbInfoRelation{
					ReferObjectId:      v.ChartInfoId,
					ReferObjectType:    utils.EDB_RELATION_CHART,
					ReferObjectSubType: chartInfo.Source,
					EdbInfoId:          v.EdbInfoId,
					EdbName:            edbInfo.EdbName,
					Source:             edbInfo.Source,
					EdbCode:            edbInfo.EdbCode,
					CreateTime:         nowTime,
					ModifyTime:         nowTime,
					RelationTime:       v.CreateTime,
				}
				addList = append(addList, tmp)
				existRelationMap[name] = struct{}{}
				if len(addList) > pageSize {
					err = data_manage.AddEdbInfoRelationMulti(addList)
					if err != nil {
						err = fmt.Errorf("新增引用记录失败 Err:%s", err)
						return
					}
					addNum += len(addList)
					addList = make([]*data_manage.EdbInfoRelation, 0)
				}
			}
		}
	}
	//拿到500个数据ID，判断相关的引用记录，如果已存在则直接过滤，不存在则新增
	if len(addList) > 0 {
		err = data_manage.AddEdbInfoRelationMulti(addList)
		if err != nil {
			err = fmt.Errorf("新增引用记录失败 Err:%s", err)
			return
		}
		addNum += len(addList)
	}
	fmt.Printf("图表指标引用记录处理完成, 新增%d条记录\n", addNum)
	return
}

// InitChartCrossVariety 处理特殊图表，跨品种分析图表
func InitChartCrossVariety() {
	fmt.Println("开始跨品种分析图表中的指标引用")
	var addNum int
	var err error
	defer func() {
		if err != nil {
			msg := fmt.Sprintf("初始化指标在跨品种分析图表中的引用失败 InitChartCrossVariety  err: %v", err)
			utils.FileLog.Info(msg)
			fmt.Println(msg)
			go alarm_msg.SendAlarmMsg(msg, 3)
		}
	}()
	total, err := cross_variety.GetChartInfoCrossVarietyTotal()
	if err != nil {
		err = fmt.Errorf("查询图表关联指标失败 err: %v", err)
		return
	}
	if total == 0 {
		return
	}
	//分页查询，每次处理500条记录
	pageSize := 500
	totalPage := (total + pageSize - 1) / pageSize // 使用整数除法，并添加一页以防有余数
	addList := make([]*data_manage.EdbInfoRelation, 0)
	//查询图表列表
	for i := 0; i < totalPage; i += 1 {
		startSize := i * pageSize
		list, e := cross_variety.GetChartInfoCrossVarietyList(startSize, pageSize)
		if e != nil {
			err = fmt.Errorf("查询图表关联指标列表失败 Err:%s", e)
			return
		}
		if len(list) == 0 {
			break
		}
		chartIds := make([]int, 0)
		tagIds := make([]int, 0)
		tagIdsMap := make(map[int]struct{})
		tagChartMap := make(map[int][]*cross_variety.ChartInfoCrossVariety)
		for _, v := range list {
			if _, ok := tagIdsMap[v.ChartXTagId]; !ok {
				tagIds = append(tagIds, v.ChartXTagId)
				tagIdsMap[v.ChartXTagId] = struct{}{}
			}
			if _, ok := tagIdsMap[v.ChartYTagId]; !ok {
				tagIds = append(tagIds, v.ChartYTagId)
				tagIdsMap[v.ChartYTagId] = struct{}{}
			}
			if chartCross, ok := tagChartMap[v.ChartXTagId]; ok {
				chartCross = append(chartCross, v)
				tagChartMap[v.ChartXTagId] = chartCross
			} else {
				chartCross = make([]*cross_variety.ChartInfoCrossVariety, 0)
				chartCross = append(chartCross, v)
				tagChartMap[v.ChartXTagId] = chartCross
			}
			if chartCross, ok := tagChartMap[v.ChartYTagId]; ok {
				chartCross = append(chartCross, v)
				tagChartMap[v.ChartYTagId] = chartCross
			} else {
				chartCross = make([]*cross_variety.ChartInfoCrossVariety, 0)
				chartCross = append(chartCross, v)
				tagChartMap[v.ChartYTagId] = chartCross
			}
		}
		chartTagVarietyList, e := cross_variety.GetChartTagVarietyEdbInfoIdsByTagIds(tagIds)
		if e != nil {
			err = fmt.Errorf("查询指标信息列表失败 Err:%s", e)
			return
		}
		edbInfoIds := make([]int, 0)
		chartTagVarietyMap := make(map[int][]*cross_variety.ChartTagVariety)
		for _, v := range chartTagVarietyList {
			if tagList, ok := chartTagVarietyMap[v.EdbInfoId]; ok {
				tagList = append(tagList, v)
				chartTagVarietyMap[v.EdbInfoId] = tagList
			} else {
				tagList = make([]*cross_variety.ChartTagVariety, 0)
				tagList = append(tagList, v)
				chartTagVarietyMap[v.EdbInfoId] = tagList
			}

			edbInfoIds = append(edbInfoIds, v.EdbInfoId)
		}

		// 查询指标信息表
		edbInfoList, e := data_manage.GetEdbInfoByIdListAndSource(edbInfoIds, []int{utils.DATA_SOURCE_WIND, utils.DATA_SOURCE_MYSTEEL_CHEMICAL})
		if e != nil {
			err = fmt.Errorf("查询指标信息列表失败 Err:%s", e)
			return
		}
		if len(edbInfoList) == 0 {
			continue
		}
		edbInfoMap := make(map[int]*data_manage.EdbInfo)
		chartInfoCrossMap := make(map[int]struct{})
		chartInfoCrossList := make([]*cross_variety.ChartInfoCrossVariety, 0)
		edbCrossMap := make(map[int][]*cross_variety.ChartInfoCrossVariety)
		for _, v := range edbInfoList {
			edbInfoMap[v.EdbInfoId] = v
			if tagList, ok := chartTagVarietyMap[v.EdbInfoId]; ok {
				for _, tag := range tagList {
					if chartCross, ok2 := tagChartMap[tag.ChartTagId]; ok2 {
						for _, crossItem := range chartCross {
							if _, ok3 := chartInfoCrossMap[crossItem.ChartInfoId]; !ok3 {
								chartInfoCrossMap[crossItem.ChartInfoId] = struct{}{}
								chartInfoCrossList = append(chartInfoCrossList, crossItem)
								chartIds = append(chartIds, crossItem.ChartInfoId)
							}
						}
					}
				}
			}
			edbCrossMap[v.EdbInfoId] = chartInfoCrossList
			chartInfoCrossMap = make(map[int]struct{})
			chartInfoCrossList = make([]*cross_variety.ChartInfoCrossVariety, 0)
		}

		/*// 查询图表信息
		chartInfoList, e := data_manage.GetChartInfoByChartInfoIds(chartIds)
		if e != nil {
			err = fmt.Errorf("查询图表信息列表失败 Err:%s", e)
			return
		}
		chartInfoMap := make(map[int]*data_manage.ChartInfo)
		for _, v := range chartInfoList {
			chartInfoMap[v.ChartInfoId] = v
		}
		*/
		//查询引用关系列表，
		chartEdbRelationList, e := data_manage.GetEdbInfoRelationByReferObjectIds(chartIds, utils.EDB_RELATION_CHART)
		if e != nil {
			err = fmt.Errorf("查询图表引用关系列表失败 Err:%s", e)
			return
		}
		existRelationMap := make(map[string]struct{})
		for _, v := range chartEdbRelationList {
			name := fmt.Sprintf("%d-%d", v.ReferObjectId, v.EdbInfoId)
			existRelationMap[name] = struct{}{}
		}
		for edbInfoId, chartCrossList := range edbCrossMap {
			nowTime := time.Now()
			for _, item := range chartCrossList {
				name := fmt.Sprintf("%d-%d", item.ChartInfoId, edbInfoId)
				if _, ok1 := existRelationMap[name]; !ok1 {
					edbInfo, ok2 := edbInfoMap[edbInfoId]
					if !ok2 {
						continue
					}
					//todo 引用时间
					tmp := &data_manage.EdbInfoRelation{
						ReferObjectId:      item.ChartInfoId,
						ReferObjectType:    utils.EDB_RELATION_CHART,
						ReferObjectSubType: utils.CHART_SOURCE_CROSS_HEDGING,
						EdbInfoId:          edbInfoId,
						EdbName:            edbInfo.EdbName,
						Source:             edbInfo.Source,
						EdbCode:            edbInfo.EdbCode,
						CreateTime:         nowTime,
						ModifyTime:         nowTime,
						RelationTime:       item.CreateTime,
					}
					addList = append(addList, tmp)
					existRelationMap[name] = struct{}{}
					if len(addList) > pageSize {
						err = data_manage.AddEdbInfoRelationMulti(addList)
						if err != nil {
							err = fmt.Errorf("新增引用记录失败 Err:%s", err)
							return
						}
						addNum += len(addList)
						addList = make([]*data_manage.EdbInfoRelation, 0)
					}
				}
			}
		}
	}
	//拿到500个数据ID，判断相关的引用记录，如果已存在则直接过滤，不存在则新增
	if len(addList) > 0 {
		err = data_manage.AddEdbInfoRelationMulti(addList)
		if err != nil {
			err = fmt.Errorf("新增引用记录失败 Err:%s", err)
			return
		}
		addNum += len(addList)
	}
	fmt.Printf("跨品种分析图表指标引用记录处理完成, 新增%d条记录\n", addNum)
	return
}

// 初始化事件日历中的指标引用
func initCalendarIndicatorRelation() {
	fmt.Println("开始处理事件日历中的指标引用")
	var addNum int
	var err error
	defer func() {
		if err != nil {
			msg := fmt.Sprintf("初始化指标在事件日历中的引用失败 initCalendarIndicatorRelation  err: %v", err)
			utils.FileLog.Info(msg)
			fmt.Println(msg)
			go alarm_msg.SendAlarmMsg(msg, 3)
		}
	}()
	//查询chart_edb_mapping 表
	obj := new(fe_calendar.FeCalendarMatter)
	condition := " AND edb_info_id > 0"
	total, err := obj.GetCountByCondition(condition, []interface{}{})
	if err != nil {
		err = fmt.Errorf("查询事件日历关联指标失败 err: %v", err)
		return
	}
	if total == 0 {
		return
	}
	//分页查询，每次处理500条记录
	pageSize := 500
	totalPage := (total + pageSize - 1) / pageSize // 使用整数除法，并添加一页以防有余数
	addList := make([]*data_manage.EdbInfoRelation, 0)
	//查询图表列表
	for i := 0; i < totalPage; i += 1 {
		startSize := i * pageSize
		list, e := obj.GetPageItemsByCondition(condition, []interface{}{}, []string{}, "", startSize, pageSize)
		if e != nil {
			err = fmt.Errorf("查询事件日历关联指标列表失败 Err:%s", e)
			return
		}
		if len(list) == 0 {
			break
		}

		edbInfoIds := make([]int, 0)
		edbInfoMatterMap := make(map[int][]*fe_calendar.FeCalendarMatter)
		for _, v := range list {
			edbInfoIds = append(edbInfoIds, v.EdbInfoId)
			items, ok := edbInfoMatterMap[v.EdbInfoId]
			if ok {
				items = append(items, v)
				edbInfoMatterMap[v.EdbInfoId] = items
			} else {
				items = make([]*fe_calendar.FeCalendarMatter, 0)
				items = append(items, v)
				edbInfoMatterMap[v.EdbInfoId] = items
			}
		}
		// 查询指标信息表
		edbInfoList, e := data_manage.GetEdbInfoByIdListAndSource(edbInfoIds, []int{utils.DATA_SOURCE_WIND, utils.DATA_SOURCE_MYSTEEL_CHEMICAL})
		if e != nil {
			err = fmt.Errorf("查询指标信息列表失败 Err:%s", e)
			return
		}
		if len(edbInfoList) == 0 {
			continue
		}
		edbInfoMap := make(map[int]*data_manage.EdbInfo)
		matterIds := make([]int, 0)
		for _, v := range edbInfoList {
			edbInfoMap[v.EdbInfoId] = v
			items, ok := edbInfoMatterMap[v.EdbInfoId]
			if ok {
				for _, item := range items {
					matterIds = append(matterIds, item.FeCalendarMatterId)
				}
			}
		}

		//查询引用关系列表，
		chartEdbRelationList, e := data_manage.GetEdbInfoRelationByReferObjectIds(matterIds, utils.EDB_RELATION_CALENDAR)
		if e != nil {
			err = fmt.Errorf("查询图表引用关系列表失败 Err:%s", e)
			return
		}
		existRelationMap := make(map[string]struct{})
		for _, v := range chartEdbRelationList {
			name := fmt.Sprintf("%d-%d", v.ReferObjectId, v.EdbInfoId)
			existRelationMap[name] = struct{}{}
		}
		for edbInfoId, edbInfo := range edbInfoMap {
			nowTime := time.Now()
			items, ok := edbInfoMatterMap[edbInfoId]
			if ok {
				for _, v := range items {
					name := fmt.Sprintf("%d-%d", v.FeCalendarMatterId, v.EdbInfoId)
					if _, ok1 := existRelationMap[name]; !ok1 {
						//todo 引用时间
						tmp := &data_manage.EdbInfoRelation{
							ReferObjectId:   v.FeCalendarMatterId,
							ReferObjectType: utils.EDB_RELATION_CALENDAR,
							EdbInfoId:       v.EdbInfoId,
							EdbName:         edbInfo.EdbName,
							Source:          edbInfo.Source,
							EdbCode:         edbInfo.EdbCode,
							CreateTime:      nowTime,
							ModifyTime:      nowTime,
							RelationTime:    v.CreateTime,
						}
						addList = append(addList, tmp)
						existRelationMap[name] = struct{}{}
						if len(addList) > pageSize {
							err = data_manage.AddEdbInfoRelationMulti(addList)
							if err != nil {
								err = fmt.Errorf("新增引用记录失败 Err:%s", err)
								return
							}
							addNum += len(addList)
							addList = make([]*data_manage.EdbInfoRelation, 0)
						}
					}
				}
			}

		}
	}
	//拿到500个数据ID，判断相关的引用记录，如果已存在则直接过滤，不存在则新增
	if len(addList) > 0 {
		err = data_manage.AddEdbInfoRelationMulti(addList)
		if err != nil {
			err = fmt.Errorf("新增引用记录失败 Err:%s", err)
			return
		}
		addNum += len(addList)
	}
	fmt.Printf("事件日历指标引用记录处理完成, 新增%d条记录\n", addNum)
	return
}

// 处理逻辑图中的指标引用
func initSandBoxEdbRelation() {
	fmt.Println("开始处理逻辑图中的指标引用")
	var err error
	var addNum int
	defer func() {
		if err != nil {
			msg := fmt.Sprintf("初始化指标在逻辑图中的引用失败 initSandBoxEdbRelation  err: %v", err)
			utils.FileLog.Info(msg)
			fmt.Println(msg)
			go alarm_msg.SendAlarmMsg(msg, 3)
		}
	}()
	condition := " AND is_delete = 0"
	total, err := sandbox.GetSandboxListCountByCondition(condition, []interface{}{})
	if err != nil {
		err = fmt.Errorf("查询逻辑图总数失败 err: %v", err)
		return
	}
	if total == 0 {
		return
	}
	//分页查询，每次处理500条记录
	pageSize := 100
	totalPage := (total + pageSize - 1) / pageSize // 使用整数除法，并添加一页以防有余数
	addList := make([]*data_manage.EdbInfoRelation, 0)
	//查询图表列表
	for i := 0; i < totalPage; i += 1 {
		startSize := i * pageSize
		list, e := sandbox.GetSandboxListByCondition(condition, []interface{}{}, startSize, pageSize)
		if e != nil {
			err = fmt.Errorf("查询逻辑图列表失败 Err:%s", e)
			return
		}
		if len(list) == 0 {
			break
		}

		edbInfoIds := make([]int, 0)
		edbSandboxMap := make(map[int][]*sandbox.Sandbox)
		for _, v := range list {
			if v.Content == "" {
				continue
			}
			edbInfoIdsTmp, e := getSandBoxEdbIdsByContent(v.Content)
			if e != nil {
				continue
				//err = fmt.Errorf("查询逻辑图关联的指标Id失败 Err:%s", e)
				//return
			}
			for _, edbId := range edbInfoIdsTmp {
				edbInfoIds = append(edbInfoIds, edbId)
				edbSandboxMap[edbId] = append(edbSandboxMap[edbId], v)
			}
		}
		if len(edbInfoIds) <= 0 {
			continue
		}
		// 查询指标信息表
		edbInfoList, e := data_manage.GetEdbInfoByIdListAndSource(edbInfoIds, []int{utils.DATA_SOURCE_WIND, utils.DATA_SOURCE_MYSTEEL_CHEMICAL})
		if e != nil {
			err = fmt.Errorf("查询指标信息列表失败 Err:%s", e)
			return
		}
		if len(edbInfoList) == 0 {
			continue
		}
		edbInfoMap := make(map[int]*data_manage.EdbInfo)
		sandboxIds := make([]int, 0)
		for _, v := range edbInfoList {
			edbInfoMap[v.EdbInfoId] = v
			if items, ok := edbSandboxMap[v.EdbInfoId]; ok {
				for _, item := range items {
					sandboxIds = append(sandboxIds, item.SandboxId)
				}
			}
		}

		//查询引用关系列表，
		chartEdbRelationList, e := data_manage.GetEdbInfoRelationByReferObjectIds(sandboxIds, utils.EDB_RELATION_SANDBOX)
		if e != nil {
			err = fmt.Errorf("查询逻辑图引用关系列表失败 Err:%s", e)
			return
		}
		existRelationMap := make(map[string]struct{})
		for _, v := range chartEdbRelationList {
			name := fmt.Sprintf("%d-%d", v.ReferObjectId, v.EdbInfoId)
			existRelationMap[name] = struct{}{}
		}
		for edbInfoId, sandboxList := range edbSandboxMap {
			nowTime := time.Now()
			for _, v := range sandboxList {
				name := fmt.Sprintf("%d-%d", v.SandboxId, edbInfoId)
				if _, ok := existRelationMap[name]; !ok {
					edbInfo, ok2 := edbInfoMap[edbInfoId]
					if !ok2 {
						continue
					}
					//todo 引用时间
					tmp := &data_manage.EdbInfoRelation{
						ReferObjectId:   v.SandboxId,
						ReferObjectType: utils.EDB_RELATION_SANDBOX,
						EdbInfoId:       edbInfoId,
						EdbName:         edbInfo.EdbName,
						Source:          edbInfo.Source,
						EdbCode:         edbInfo.EdbCode,
						CreateTime:      nowTime,
						ModifyTime:      nowTime,
						RelationTime:    v.CreateTime,
					}
					addList = append(addList, tmp)
					existRelationMap[name] = struct{}{}
					if len(addList) > pageSize {
						err = data_manage.AddEdbInfoRelationMulti(addList)
						if err != nil {
							err = fmt.Errorf("新增引用记录失败 Err:%s", err)
							return
						}
						addNum += len(addList)
						addList = make([]*data_manage.EdbInfoRelation, 0)
					}
				}
			}
		}
	}
	//拿到500个数据ID，判断相关的引用记录，如果已存在则直接过滤，不存在则新增
	if len(addList) > 0 {
		err = data_manage.AddEdbInfoRelationMulti(addList)
		if err != nil {
			err = fmt.Errorf("新增引用记录失败 Err:%s", err)
			return
		}
		addNum += len(addList)
	}
	fmt.Printf("逻辑图指标引用记录处理完成, 新增%d条记录\n", addNum)
	return
}

func getSandBoxEdbIdsByContent(content string) (edbInfoIds []int, err error) {
	var contentInfo sandbox.ContentDataStruct
	err = json.Unmarshal([]byte(content), &contentInfo)
	if err != nil {
		err = fmt.Errorf("json.Unmarshal err:%s", err.Error())
		return
	}
	// 遍历所有节点
	for _, node := range contentInfo.Cells {
		if node.Data == nil {
			continue
		}
		for _, v := range node.Data.LinkData {
			if v.Type == 1 {
				edbInfoIds = append(edbInfoIds, v.Id)
			}
		}
	}
	return
}