睿姿得数据爬取

models/base_from_rzd_data.go

@@ -2,11 +2,11 @@
 package models
 
 type BaseFromRzdData struct {
-	BaseFromLyDataId  int     `orm:"column(base_from_ly_data_id);pk"`
-	BaseFromLyIndexId int     `orm:"column(base_from_ly_index_id)"`
-	CreateTime        string  `orm:"column(create_time)"`
-	DataTime          string  `orm:"column(data_time)"`
-	IndexCode         string  `orm:"column(index_code)"`
-	ModifyTime        string  `orm:"column(modify_time)"`
-	Value             float64 `orm:"column(value)"`
+	BaseFromRzdDataId  int     `orm:"column(base_from_rzd_data_id);pk"`
+	BaseFromRzdIndexId int     `orm:"column(base_from_rzd_index_id)"`
+	CreateTime         string  `orm:"column(create_time)"`
+	DataTime           string  `orm:"column(data_time)"`
+	IndexCode          string  `orm:"column(index_code)"`
+	ModifyTime         string  `orm:"column(modify_time)"`
+	Value              float64 `orm:"column(value)"`
 }

models/edb_data_rzd.go

@@ -0,0 +1,14 @@
+// Package models
+// @Author gmy 2024/9/14 16:13:00
+package models
+
+type EdbDataRzd struct {
+	edbDataId     int     `orm:"column(edb_data_id);pk"`
+	CreateTime    string  `orm:"column(create_time)"`
+	ModifyTime    string  `orm:"column(modify_time)"`
+	EdbInfoId     int     `orm:"column(edb_info_id)"`
+	EdbCode       string  `orm:"column(edb_code)"`
+	DataTime      string  `orm:"column(data_time)"`
+	Value         float64 `orm:"column(value)"`
+	DataTimestamp uint64  `orm:"column(data_timestamp)"`
+}

services/ruizide/data_processor.go

@@ -2,6 +2,8 @@ package ruizide
 
 import (
 	"context"
+	"encoding/json"
+	"eta/eta_data_analysis/models"
 	"eta/eta_data_analysis/utils"
 	"fmt"
 	"github.com/chromedp/cdproto/cdp"
@@ -337,6 +339,7 @@ func main() {
 		}
 
 		// 遍历行并打印内容
+		indexData := []models.BaseFromRzdData{}
 		for rowIndex, rowData := range maxRow {
 			if rowIndex < 4 {
 				continue
@@ -346,8 +349,75 @@ func main() {
 				return
 			}
 			baseFromLyDataList, err := processor.Process(tableName, sheetName, rowData)
+			indexData = append(indexData, baseFromLyDataList...)
+		}
+
+		// 新增数据源指标数据
+		if len(indexData) > 0 {
+			// 转换成json
+			marshal, err := json.Marshal(indexData)
+			if err != nil {
+				log.Printf("postEdbLib err: %v", err)
+				return
+			}
+			_, err = utils.HttpPostRequest(utils.EDB_LIB_URL+utils.ADD_RZD_DATA_LIST, string(marshal), "application/json")
+			if err != nil {
+				// 有错误就不继续执行
+				log.Printf("postEdbLib err: %v", err)
+				return
+			}
+		}
+
+		// 新增指标库数据
+		edbDataList := []models.EdbDataRzd{}
+		for _, index := range indexData {
+			// 补充 判断是否存在于指标库
+			paramsLib := make(map[string]interface{})
+			paramsLib["IndexCode"] = index.IndexCode
+			paramsLib["DataTime"] = index.DataTime
+			postEdbLib, err := httpRequestFill(paramsLib, utils.GET_EDB_DATA_BY_INDEX_CODE)
+			if err != nil {
+				// 有错误就不继续执行
+				log.Printf("postEdbLib err: %v", err)
+				continue
+			}
+			var requestResponse models.RequestResponse[models.EdbInfo]
+			err = json.Unmarshal(postEdbLib, &requestResponse)
+			if err != nil {
+				log.Printf("postEdbLib err: %v", err)
+				continue
+			}
+
+			if requestResponse.Data.EdbInfoId == 0 {
+				edbDataRzd := models.EdbDataRzd{
+					CreateTime:    utils.GetCurrentTime(),
+					ModifyTime:    utils.GetCurrentTime(),
+					EdbInfoId:     index.BaseFromRzdIndexId,
+					EdbCode:       index.IndexCode,
+					DataTime:      index.DataTime,
+					Value:         index.Value,
+					DataTimestamp: uint64(time.Now().UnixNano() / int64(time.Millisecond)),
+				}
+				edbDataList = append(edbDataList, edbDataRzd)
+			}
 
 		}
+
+		if len(edbDataList) > 0 {
+			// 转换成json
+			marshal, err := json.Marshal(edbDataList)
+			if err != nil {
+				log.Printf("postEdbLib err: %v", err)
+				return
+			}
+			_, err = utils.HttpPostRequest(utils.EDB_LIB_URL+utils.ADD_BATCH_RZD_EDB_DATA, string(marshal), "application/json")
+			if err != nil {
+				// 有错误就不继续执行
+				log.Printf("postEdbLib err: %v", err)
+				return
+			}
+		}
+
 	}
 }
 
@@ -389,3 +459,25 @@ func login(ctx context.Context) error {
 		chromedp.Sleep(5*time.Second),                                 // 等待页面加载完成
 	)
 }
+
+func httpRequestFill(data interface{}, urlMethod string) (postEdbLib []byte, err error) {
+	// 转换成json
+	marshal, err := json.Marshal(data)
+	if err != nil {
+		return nil, err
+	}
+	// json 转 interface
+	var result map[string]interface{}
+	err = json.Unmarshal(marshal, &result)
+	if err != nil {
+		return nil, err
+	}
+
+	postEdbLib, err = utils.PostEdbLibRequest(result, urlMethod)
+	if err != nil {
+		// 有错误就不继续执行
+		log.Printf("postEdbLib err: %v", err)
+		return nil, err
+	}
+	return postEdbLib, nil
+}

services/ruizide/processor_business_logic.go

@@ -3,8 +3,14 @@
 package ruizide
 
 import (
+	"encoding/json"
 	"eta/eta_data_analysis/models"
+	"eta/eta_data_analysis/utils"
+	"fmt"
 	"github.com/beego/beego/v2/core/logs"
+	"strconv"
+	"strings"
+	"unicode"
 )
 
 var classifyMap = map[string]string{
@@ -18,19 +24,283 @@ var classifyMap = map[string]string{
 	"Demand - Maritime Bunker": "analytics library",
 }
 
-// AnalyticsLibraryProcessor
+// RoadIndexProcessor
 // @Description: AnalyticsLibrary处理器
-type AnalyticsLibraryProcessor struct{}
+type RoadIndexProcessor struct{}
 
-func (p *AnalyticsLibraryProcessor) Process(tableName string, sheetName string, rowData []string) ([]models.BaseFromRzdData, error) {
+func (p *RoadIndexProcessor) Process(tableName string, sheetName string, rowData []string) ([]models.BaseFromRzdData, error) {
 	logs.Info("Processing AnalyticsLibrary...")
+	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)
 
-	return nil, nil
+	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
+	}
+	dataTimeOne := rowData[1]
+
+	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
+	}
+	dataTimeTwo := rowData[1]
+
+	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, rowData[len(rowData)-3], indexNameColOne)
+	indexCodeTwo, err = getIndexId(sheetName, 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"] = 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["classifyName"] = 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
 }

services/ruizide/processor_factory.go

@@ -12,13 +12,15 @@ type ReportProcessor interface {
 func GetProcessor(tableName string, sheetName string) (ReportProcessor, error) {
 	if tableName == "Oil_Demand_Signals_Weekly_Report" {
 		switch sheetName {
+		case "Road Index":
+			return &RoadIndexProcessor{}, nil
 		default:
-			return &AnalyticsLibraryProcessor{}, nil
+			return nil, fmt.Errorf("unknown sheetName: %s", sheetName)
 		}
 	} else if tableName == "豆粕" {
 		switch sheetName {
 		default:
-			return nil, fmt.Errorf("unknown category: %s", sheetName)
+			return nil, fmt.Errorf("unknown sheetName: %s", sheetName)
 		}
 	}
 	// 可以添加更多的逻辑来处理其他产品和类别

utils/constants.go

@@ -98,6 +98,7 @@ const (
 	DATA_SOURCE_PREDICT_CALCULATE_PERCENTILE                    //预测百分位->70
 	DATA_SOURCE_FUBAO                                           //富宝数据->71
 	DATA_SOURCE_LY                                   = 91       // 粮油商务网
+	DATA_SOURCE_RZD                                  = 97       // 睿姿得数据
 )
 
 // 指标来源的中文展示