eta_data_analysis/services/ruizide/processor_business_logic.go

// Package ruizide
// @Author gmy 2024/10/21 10:50:00
package main

import (
	"encoding/json"
	"eta/eta_data_analysis/models"
	"eta/eta_data_analysis/utils"
	"fmt"
	"github.com/beego/beego/v2/core/logs"
	"math"
	"strconv"
	"strings"
	"unicode"
)

var classifyMap = map[string]string{
	"Road Index":                       "analytics library",
	"Road Active Fleet":                "analytics library",
	"Aviation Index":                   "analytics library",
	"Aviation Active Fleet":            "analytics library",
	"Demand-Gasoline":                  "analytics library",
	"Demand - Diesel":                  "analytics library",
	"Demand - Jet Fuel":                "analytics library",
	"Demand - Maritime Bunker":         "analytics library",
	"Oil_Demand_Signals_Weekly_Report": "analytics library",
}

// RoadIndexProcessor
// @Description: AnalyticsLibrary处理器
type RoadIndexProcessor struct{}

func (p *RoadIndexProcessor) Process(tableName string, sheetName string, rowIndex int, rowData []string) ([]models.BaseFromRzdData, error) {
	logs.Info("Processing AnalyticsLibrary...")
	if rowIndex < 4 {
		return nil, nil
	}

	frequency := "日度"
	unit := "%"
	indexNameColOne := "Index"
	indexNameColTwo := "Index 7DMA"

	// step_1: 分类
	classifyId, err := dealClassify(tableName, sheetName)
	if err != nil {
		return nil, err
	}
	logs.Info("classifyId: %v", classifyId)

	// step_2: 指标
	indexOneId, indexTwoId, indexCodeOne, indexCodeTwo, err := dealIndex(sheetName, rowData, indexNameColOne, indexNameColTwo, frequency, unit, classifyId)
	if err != nil {
		return nil, err
	}
	logs.Info("indexOneId: %v, indexTwoId: %v, indexCodeOne: %v, indexCodeTwo: %v", indexOneId, indexTwoId, indexCodeOne, indexCodeTwo)

	// step_3: 指标数据
	dataList, err := dealData(indexOneId, indexTwoId, indexCodeOne, indexCodeTwo, rowData)
	if err != nil {
		return nil, err
	}
	logs.Info("dataList: %v", dataList)

	return dataList, err
}

// ReDashboardExportOneProcessor
// @Description: ReDashboardExportOneProcessor处理器
type ReDashboardExportOneProcessor struct{}

func (p *ReDashboardExportOneProcessor) Process(tableName string, sheetName string, rowIndex int, rowData []string) ([]models.BaseFromRzdData, error) {
	logs.Info("Processing ReDashboardExportOne...")
	if rowIndex < 4 {
		return nil, nil
	}

	frequency := "季度"
	unit := "千桶每天"
	indexNameColOne := "Index"
	indexNameColTwo := "Index 7DMA"

	// step_1: 分类
	classifyId, err := dealClassify("cube dashboards", "Supply Revision Analysis")
	if err != nil {
		return nil, err
	}
	logs.Info("classifyId: %v", classifyId)

	// step_2: 指标
	indexOneId, indexTwoId, indexCodeOne, indexCodeTwo, err := dealIndex(sheetName, rowData, indexNameColOne, indexNameColTwo, frequency, unit, classifyId)
	if err != nil {
		return nil, err
	}
	logs.Info("indexOneId: %v, indexTwoId: %v, indexCodeOne: %v, indexCodeTwo: %v", indexOneId, indexTwoId, indexCodeOne, indexCodeTwo)

	// step_3: 指标数据
	dataList, err := dealData(indexOneId, indexTwoId, indexCodeOne, indexCodeTwo, rowData)
	if err != nil {
		return nil, err
	}
	logs.Info("dataList: %v", dataList)

	return dataList, err
}

func dealData(indexOneId, indexTwoId int, indexCodeOne, indexCodeTwo string, rowData []string) ([]models.BaseFromRzdData, error) {
	var dataList []models.BaseFromRzdData

	valueOne, err := strconv.ParseFloat(rowData[len(rowData)-2], 64)
	if err != nil {
		return nil, err
	}
	valueOne = math.Round(valueOne*10000) / 10000

	dataTimeOne := rowData[1]
	formatOne, err := utils.ConvertDateFormat(dataTimeOne)
	if err != nil {
		return nil, err
	}
	dataTimeOne = formatOne

	paramsLib := make(map[string]interface{})
	paramsLib["IndexCode"] = indexCodeOne
	paramsLib["DataTime"] = dataTimeOne
	postEdbLib, err := httpRequestFill(paramsLib, utils.GET_RZD_INDEX_DATA_BY_CODE_AND_TIME)
	if err != nil {
		return nil, err
	}
	var requestResponse models.RequestResponse[models.BaseFromRzdData]
	err = json.Unmarshal(postEdbLib, &requestResponse)
	if err != nil {
		return nil, err
	}
	if requestResponse.Data.BaseFromRzdIndexId == 0 {
		dataOne := models.BaseFromRzdData{
			BaseFromRzdIndexId: indexOneId,
			CreateTime:         utils.GetCurrentTime(),
			DataTime:           dataTimeOne,
			IndexCode:          indexCodeOne,
			ModifyTime:         utils.GetCurrentTime(),
			Value:              valueOne,
		}
		dataList = append(dataList, dataOne)
	}

	valueTwo, err := strconv.ParseFloat(rowData[len(rowData)-2], 64)
	if err != nil {
		return nil, err
	}
	valueTwo = math.Round(valueTwo*10000) / 10000

	dataTimeTwo := rowData[1]
	formatTwo, err := utils.ConvertDateFormat(dataTimeTwo)
	if err != nil {
		return nil, err
	}
	dataTimeTwo = formatTwo

	paramsLib = make(map[string]interface{})
	paramsLib["indexCode"] = indexCodeTwo
	paramsLib["dataTime"] = dataTimeTwo
	postEdbLib, err = httpRequestFill(paramsLib, utils.GET_RZD_INDEX_DATA_BY_CODE_AND_TIME)
	if err != nil {
		return nil, err
	}
	var requestResponseTwo models.RequestResponse[models.BaseFromRzdData]
	err = json.Unmarshal(postEdbLib, &requestResponseTwo)
	if err != nil {
		return nil, err
	}
	if requestResponseTwo.Data.BaseFromRzdIndexId == 0 {
		dataTwo := models.BaseFromRzdData{
			BaseFromRzdIndexId: indexTwoId,
			CreateTime:         utils.GetCurrentTime(),
			DataTime:           dataTimeTwo,
			IndexCode:          indexCodeTwo,
			ModifyTime:         utils.GetCurrentTime(),
			Value:              valueTwo,
		}
		dataList = append(dataList, dataTwo)
	}

	return dataList, nil
}

func dealIndex(sheetName string, rowData []string, indexNameColOne string, indexNameColTwo string, frequency string, unit string, classifyId int) (indexOneId, indexTwoId int, indexCodeOne, indexCodeTwo string, err error) {
	// 指标名称
	indexNameOne := sheetName + "/" + rowData[len(rowData)-3] + "/" + indexNameColOne
	indexNameTwo := sheetName + "/" + rowData[len(rowData)-3] + "/" + indexNameColTwo
	// 生成指标编码
	indexCodeOne, err = getIndexId(sheetName, strings.ToLower(rowData[len(rowData)-3]), indexNameColOne)
	indexCodeTwo, err = getIndexId(sheetName, strings.ToLower(rowData[len(rowData)-3]), indexNameColTwo)

	// 处理第一个指标
	paramsLib := make(map[string]interface{})
	paramsLib["indexCode"] = indexCodeOne
	postEdbLib, err := httpRequestFill(paramsLib, utils.GET_RZD_INDEX_BY_CODE)
	if err != nil {
		return 0, 0, "0", "0", fmt.Errorf("getIndexId() : Failed to get rzd index by code: %v", err)
	}
	var requestResponse models.RequestResponse[models.BaseFromRzdIndex]
	err = json.Unmarshal(postEdbLib, &requestResponse)
	if err != nil {
		return 0, 0, "0", "0", err
	}
	if requestResponse.Data.BaseFromRzdIndexId == 0 {
		indexOne := models.BaseFromRzdIndex{
			CreateTime:           utils.GetCurrentTime(),
			ModifyTime:           utils.GetCurrentTime(),
			BaseFromLyClassifyId: classifyId,
			IndexCode:            indexCodeOne,
			IndexName:            indexNameOne,
			Frequency:            frequency,
			Unit:                 unit,
		}
		// 这里避免服务器宕机 出现唯一索引异常，进行分开保存
		postEdbLib, err = httpRequestFill(indexOne, utils.ADD_RZD_INDEX)
		if err != nil {
			return 0, 0, "0", "0", fmt.Errorf("getIndexId() : Failed to get rzd index by code: %v", err)
		}
		var requestResponse models.RequestResponse[int]
		err = json.Unmarshal(postEdbLib, &requestResponse)
		if err != nil {
			return 0, 0, "0", "0", err
		}
		indexOneId = requestResponse.Data
		logs.Info("indexOneId: %v", indexOneId)

	} else {
		indexOneId = requestResponse.Data.BaseFromRzdIndexId
	}

	// 处理第二个指标
	paramsLib = make(map[string]interface{})
	paramsLib["indexCode"] = indexCodeTwo
	postEdbLib, err = httpRequestFill(paramsLib, utils.GET_RZD_INDEX_BY_CODE)
	if err != nil {
		return 0, 0, "0", "0", fmt.Errorf("getIndexId() : Failed to get rzd index by code: %v", err)
	}
	var requestResponseTwo models.RequestResponse[models.BaseFromRzdIndex]
	err = json.Unmarshal(postEdbLib, &requestResponseTwo)
	if err != nil {
		return 0, 0, "0", "0", err
	}
	if requestResponseTwo.Data.BaseFromRzdIndexId == 0 {
		indexTwo := models.BaseFromRzdIndex{
			CreateTime:           utils.GetCurrentTime(),
			ModifyTime:           utils.GetCurrentTime(),
			BaseFromLyClassifyId: classifyId,
			IndexCode:            indexCodeTwo,
			IndexName:            indexNameTwo,
			Frequency:            frequency,
			Unit:                 unit,
		}
		// 这里避免服务器宕机 出现唯一索引异常，进行分开保存
		var requestResponse models.RequestResponse[int]
		postEdbLib, err = httpRequestFill(indexTwo, utils.ADD_RZD_INDEX)
		if err != nil {
			return 0, 0, "0", "0", fmt.Errorf("getIndexId() : Failed to get rzd index by code: %v", err)
		}
		err = json.Unmarshal(postEdbLib, &requestResponse)
		if err != nil {
			return 0, 0, "0", "0", err
		}
		indexTwoId = requestResponse.Data
		logs.Info("indexTwoId: %v", indexTwoId)
	} else {
		indexTwoId = requestResponseTwo.Data.BaseFromRzdIndexId
	}

	return indexOneId, indexTwoId, indexCodeOne, indexCodeTwo, nil
}

func getIndexId(prefix string, area string, suffix string) (string, error) {
	prefixWords := strings.Fields(prefix) // 分割字符串为单词
	firstLetters := ""

	for _, word := range prefixWords {
		if len(word) > 0 {
			firstLetters += string(unicode.ToLower(rune(word[0]))) // 获取小写的第一个字母
		}
	}

	suffixWords := strings.Fields(suffix) // 分割字符串为单词
	SecondLetters := ""

	for _, word := range suffixWords {
		if len(word) > 0 {
			SecondLetters += string(unicode.ToLower(rune(word[0]))) // 获取小写的第一个字母
		}
	}
	indexCode := "rzd" + firstLetters + area + SecondLetters

	return indexCode, nil
}

// 处理分类
func dealClassify(tableName, sheetName string) (int, error) {
	// 查询一级分类是否存在
	paramsLib := make(map[string]interface{})
	paramsLib["classifyName"] = classifyMap[tableName]
	postEdbLib, err := httpRequestFill(paramsLib, utils.GET_RZD_CLASSIFY_BY_NAME)
	if err != nil {
		return 0, fmt.Errorf("AnalyticsLibraryProcessor Process() : failed to get classify: %v", err)
	}
	var requestResponse models.RequestResponse[models.BaseFromRzdClassify]
	err = json.Unmarshal(postEdbLib, &requestResponse)
	if err != nil {
		return 0, err
	}

	// 处理一级分类
	var parentId int
	if requestResponse.Data.BaseFromRzdClassifyId == 0 {
		// 一级分类不存在，新增一级分类
		paramsLib = make(map[string]interface{})
		paramsLib["parentId"] = 0
		paramsLib["classifyName"] = classifyMap[tableName]
		postEdbLib, err = httpRequestFill(paramsLib, utils.ADD_RZD_CLASSIFY)
		if err != nil {
			return 0, fmt.Errorf("AnalyticsLibraryProcessor Process() : failed to add classify: %v", err)
		}
		var requestResponse models.RequestResponse[int]
		err = json.Unmarshal(postEdbLib, &requestResponse)
		if err != nil {
			return 0, err
		}
		parentId = requestResponse.Data
	} else {
		// 一级分类已存在，使用其 ID
		parentId = requestResponse.Data.BaseFromRzdClassifyId
	}

	// 查询二级分类是否存在
	paramsSubLib := make(map[string]interface{})
	paramsSubLib["classifyName"] = sheetName // 这里替换成实际的二级分类名称
	postSubEdbLib, err := httpRequestFill(paramsSubLib, utils.GET_RZD_CLASSIFY_BY_NAME)
	if err != nil {
		return 0, fmt.Errorf("AnalyticsLibraryProcessor Process() : failed to get sub classify: %v", err)
	}

	var subRequestResponse models.RequestResponse[models.BaseFromRzdClassify]
	err = json.Unmarshal(postSubEdbLib, &subRequestResponse)
	if err != nil {
		return 0, err
	}

	// 新增二级分类
	var classifyId int
	if subRequestResponse.Data.BaseFromRzdClassifyId == 0 {
		paramsLib = make(map[string]interface{})
		paramsLib["parentId"] = parentId
		paramsLib["classifyName"] = sheetName
		postEdbLib, err = httpRequestFill(paramsLib, utils.ADD_RZD_CLASSIFY)
		if err != nil {
			return 0, fmt.Errorf("AnalyticsLibraryProcessor Process() : failed to add classify: %v", err)
		}
		var requestResponse models.RequestResponse[int]
		err = json.Unmarshal(postEdbLib, &requestResponse)
		if err != nil {
			return 0, err
		}
		classifyId = requestResponse.Data
	} else {
		classifyId = subRequestResponse.Data.BaseFromRzdClassifyId
	}

	return classifyId, nil
}