eta_data_analysis/services/ruizide/data_processor.go

package ruizide

import (
	"context"
	"encoding/json"
	"eta/eta_data_analysis/models"
	"eta/eta_data_analysis/utils"
	"fmt"
	"github.com/chromedp/cdproto/cdp"
	"github.com/xuri/excelize/v2"
	"io"
	"log"
	"os"
	"path/filepath"
	"time"

	"github.com/chromedp/chromedp"
)

// 定义选择器
var (
	downloadDir  = utils.RZD_EXCEL_PATH
	defaultDir   = utils.RZD_DOWNLOAD_PATH
	rzdLoginPath = "https://clients.rystadenergy.com/clients/"
	rzdBatchSize = 500

	clientSearchLink       = `div.d-none.d-lg-flex.flex-grow-1 a[href="/clients/search/"]`
	downloadButtonSelector = `div.btn.btn-link.btn-sm.dashboard-action.dashboard-action--download-data`
)

// 处理数据下载的步骤
func downloadData(ctx context.Context) error {
	// Analytics Library
	if err := chromedp.Run(ctx,
		chromedp.Sleep(5*time.Second), // 考虑移除这一行，如果不必要的话
		chromedp.Navigate(rzdLoginPath),
		chromedp.WaitVisible(`div.d-none.d-lg-flex.flex-grow-1`, chromedp.ByQuery),
		chromedp.Click(clientSearchLink, chromedp.ByQuery),
		chromedp.WaitVisible(`input[class="ais-SearchBox-input rounded border py-2 px-3 shadow-sm font-size-14 w-100"]`, chromedp.ByQuery),
		chromedp.SetValue(`input[class="ais-SearchBox-input rounded border py-2 px-3 shadow-sm font-size-14 w-100"]`, "oil demand signals weekly report", chromedp.ByQuery),
		chromedp.Sleep(10*time.Second),
		//chromedp.Click(`div.ais-InfiniteHits li a:has(img[src="/Static/img/icons/xls.png"])`, chromedp.ByQuery),
	); err != nil {
		return fmt.Errorf("下载 Analytics Library 数据错误: %v", err)
	}

	xpath := `//div[@id='search-page-hits']//li//a[.//div//span[@class='align-middle' and text()='Data']]`
	var inputCount int
	var nodes []*cdp.Node // 使用 *cdp.Node
	if err := chromedp.Run(ctx,
		chromedp.ActionFunc(func(ctx context.Context) error {

			// 获取匹配的节点
			if err := chromedp.Nodes(xpath, &nodes, chromedp.BySearch).Do(ctx); err != nil {
				return fmt.Errorf("检查节点失败: %v", err)
			}

			// 获取节点数量
			inputCount = len(nodes)
			fmt.Printf("找到 %d 个匹配的元素\n", inputCount)

			if inputCount > 0 {
				// 点击第一个节点
				return chromedp.MouseClickNode(nodes[0]).Do(ctx) // 使用 []cdp.NodeID
			}
			return nil
		}),
		chromedp.Sleep(10*time.Second),
	); err != nil {
		return fmt.Errorf("下载 Analytics Library 数据错误: %v", err)
	}
	// 解析文件移动到目标目录
	if err := waitAndRenameDownloadedFile("Oil_Demand_Signals_Weekly_Report_"+utils.GetCurrentYearMonth()+".xlsx", downloadDir); err != nil {
		return err
	}

	return nil
}

// 等待下载文件并重命名
func waitAndRenameDownloadedFile(newFileName, targetDir string) error {
	// 等待一段时间以确保文件下载完成
	time.Sleep(100 * time.Second) // 可能需要根据实际情况调整

	// 查找下载目录中的文件
	files, err := filepath.Glob(filepath.Join(defaultDir, "*.xlsx"))
	if err != nil {
		return fmt.Errorf("查找文件时出错: %v", err)
	}

	// 如果没有找到文件，返回错误
	if len(files) == 0 {
		return fmt.Errorf("未找到任何下载的文件")
	}

	// 找到最新的文件
	var latestFile string
	var latestTime time.Time

	for _, file := range files {
		info, err := os.Stat(file)
		if err != nil {
			return fmt.Errorf("获取文件信息时出错: %v", err)
		}
		if info.ModTime().After(latestTime) {
			latestTime = info.ModTime()
			latestFile = file
		}
	}

	// 目标文件的完整路径
	targetFilePath := filepath.Join(targetDir, newFileName)

	// 重命名并移动到目标目录
	if latestFile != "" {
		if err := moveFile(latestFile, targetFilePath); err != nil {
			return fmt.Errorf("重命名文件时出错: %v", err)
		}
		// 打印重命名后的文件名
		fmt.Printf("文件重命名并移动到: %s\n", targetFilePath)
	}

	return nil
}

func moveFile(source, destination string) error {
	// 复制文件
	srcFile, err := os.Open(source)
	if err != nil {
		return fmt.Errorf("打开源文件时出错: %v", err)
	}
	defer srcFile.Close()

	dstFile, err := os.Create(destination)
	if err != nil {
		return fmt.Errorf("创建目标文件时出错: %v", err)
	}
	defer dstFile.Close()

	if _, err := io.Copy(dstFile, srcFile); err != nil {
		return fmt.Errorf("复制文件时出错: %v", err)
	}
	time.Sleep(60 * time.Second)

	// 删除源文件
	/*if err := os.Remove(source); err != nil {
		return fmt.Errorf("删除源文件时出错: %v", err)
	}*/

	return nil
}

// 解析网页数据，下载文件
// func main() {
func ResolverNet(cont context.Context) (err error) {
	// 创建下载目录
	if err := os.MkdirAll(downloadDir, os.ModePerm); err != nil {
		fmt.Printf("创建下载目录时出错: %v\n", err)
		return nil
	}

	// 创建 chromedp 执行上下文
	options := []chromedp.ExecAllocatorOption{
		chromedp.Flag("headless", false),
		chromedp.Flag("disable-blink-features", "AutomationControlled"),
		chromedp.UserAgent(`Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.80 Safari/537.36`),
		// 设置了但并不生效，直接从默认下载路径读取过来
		//chromedp.Flag("download.default_directory", downloadDir),
		//chromedp.Flag("download.prompt_for_download", false), // 不弹出下载对话框
		chromedp.Flag("safebrowsing.enabled", true), // 启用安全浏览
		//chromedp.UserDataDir(filepath.Join(downloadDir, "user-data")),
	}

	allocCtx, cancel := chromedp.NewExecAllocator(context.Background(), options...)
	defer cancel()
	ctx, cancel := chromedp.NewContext(allocCtx)
	defer cancel()

	// 启动 Chrome 实例
	if err := chromedp.Run(ctx); err != nil {
		fmt.Printf("启动 Chrome 实例时出错: %v\n", err)
		return nil
	}

	// 设置下载行为
	/*if err := setDownloadBehavior(ctx); err != nil {
		fmt.Printf("设置下载路径时出错: %v\n", err)
		return
	}*/

	// 登录操作
	if err := login(ctx); err != nil {
		fmt.Printf("登录错误: %v\n", err)
		return nil
	}

	// 下载数据
	if err := downloadData(ctx); err != nil {
		fmt.Printf("数据下载错误: %v\n", err)
		return nil
	}

	fmt.Println("数据下载完成")

	// 解析表格 读取数据
	fileResolver()

	return nil
}

// 解析本地文件
func fileResolver() {
	//func main() {
	var tableNameList = []string{
		"Oil_Demand_Signals_Weekly_Report",
		"Oil_Supply_Analysis",
		"Supply_Revision_Analysis",
		"Oil_Market_Cube_Upstream_Supply_Oil_Quality_API",
		"Oil_Market_Cube_Upstream_Supply_Oil_Quality_Sulphur",
		"Oil_Market_Cube_Upstream_Supply_Capacity_Capacity",
		"Oil_Market_Cube_Upstream_Supply_Production",
		"Oil_Market_Cube_Upstream_Supply_Production_Wo_Seasonality",
		"Oil_Market_Cube_Upstream_Supply_OPEC_Policy_Reference_Production",
		"Oil_Market_Cube_Upstream_Supply_OPEC_Policy_Target_Production",
		"Oil_Market_Cube_Upstream_Supply_OPEC_Policy_Target_Cut",
		"Oil_Market_Cube_Upstream_Supply_OPEC_Policy_Actual_Cut",
		"Oil_Market_Cube_Upstream_Supply_OPEC_Policy_Compliance",
		"Oil_Market_Cube_Upstream_Supply_OPEC_Policy_Production_Subject_To_Cut",
		"Oil_Market_Cube_Products_Demand_Products_Demand_Mean",
		"Oil_Market_Cube_Products_Demand_Products_Demand+Sigma",
		"Oil_Market_Cube_Products_Demand_Products_Demand-Sigma",
		"Oil_Market_Cube_Balances_Total_Liquids_Balances",
		"Oil_Market_Cube_Geography_Latitude",
		"Oil_Market_Cube_Geography_Longitude",
		"Oil_Demand_Analysis_Product_Detail",
		"Oil_Demand_Analysis_Region",
		"Oil_Demand_Analysis_Scenario",
		"Oil_Demand_Analysis_Continent",
		"Oil_Demand_Analysis_Country",
		"Oil_Demand_Analysis_Product_Category",
		"Oil_Demand_Analysis_Sector_Category",
		"Oil_Demand_Analysis_Sector_Detail",
	}
	for _, tableName := range tableNameList {
		var fileName string
		// 解析表格
		fileName = tableName + "_" + utils.GetCurrentYearMonth() + ".xlsx"
		filePath := filepath.Join(downloadDir, fileName)

		// 打开 Excel 文件
		f, err := excelize.OpenFile(filePath)
		if err != nil {
			log.Fatalf("无法打开 Excel 文件: %v", err)
		}

		// 获取所有工作表
		sheetNames := f.GetSheetList()

		for _, sheetName := range sheetNames {
			fmt.Printf("读取工作表: %s\n", sheetName)

			// 获取工作表的最大行数
			maxRow, err := f.GetRows(sheetName) // 直接获取所有行数据
			if err != nil {
				log.Fatalf("获取工作表数据时出错: %v", err)
				continue
			}

			// 遍历行并打印内容
			indexData := []models.BaseFromRzdData{}
			for rowIndex, rowData := range maxRow {

				// 因为excel文件中的sheet表格不固定 对于 Supply_Revision_Analysis, Oil_Supply_Analysis 文件 手动调整sheet表格顺序
				if tableName == "Supply_Revision_Analysis" && rowIndex == 0 {
					if rowData[0] == "YearQuarter" && rowData[1] == "Revision" && rowData[2] == "CountryRevisionGroup" {
						sheetName = "Chart1"
					}
					if rowData[0] == "YearQuarter" && rowData[1] == "Current" && rowData[2] == "Previous" {
						sheetName = "Chart2"
					}
					if rowData[0] == "Year" && rowData[1] == "Revision" && rowData[2] == "CountryRevisionGroup" {
						sheetName = "Chart3"
					}
					if rowData[0] == "Year" && rowData[1] == "Current" && rowData[2] == "Previous" {
						sheetName = "Chart4"
					}
					if rowData[0] == "Previous" && rowData[1] == "Current" && rowData[2] == "YearMonth" {
						sheetName = "Chart5"
					}
					if rowData[0] == "YearMonth" && rowData[1] == "CountryRevisionGroup" && rowData[2] == "Revision" {
						sheetName = "Chart6"
					}
				} else if tableName == "Oil_Supply_Analysis" && rowIndex == 0 {
					if rowData[0] == "Viz Date" && rowData[1] == "OilAndGasCategory" && rowData[2] == "supply_kbbld" {
						sheetName = "Chart1"
					}
					if rowData[0] == "Viz Date" && rowData[1] == "supply_kbbld" && rowData[2] == "Region" {
						sheetName = "Chart2"
					}
					if rowData[0] == "Viz Date" && rowData[1] == "CapacityDetail" && rowData[2] == "Capacity_kbbld" {
						sheetName = "Chart3"
					}
					if rowData[0] == "Viz Date" && rowData[1] == "Oil Classification Group" && rowData[2] == "supply_kbbld" {
						sheetName = "Chart4"
					}
				}

				processor, err := GetProcessor(tableName, sheetName)
				if err != nil {
					continue
				}
				baseFromRzdDataList, err := processor.Process(tableName, sheetName, rowIndex, rowData)
				if err != nil {
					log.Printf("processor.Process err: %v", err)
					return
				}
				indexData = append(indexData, baseFromRzdDataList...)
			}

			// 新增数据源指标数据
			if len(indexData) > 0 {
				for i := 0; i < len(indexData); i += rzdBatchSize {
					// 计算当前批次的结束索引
					end := i + rzdBatchSize
					if end > len(indexData) {
						end = len(indexData)
					}

					// 获取当前批次的数据
					batchData := indexData[i:end]

					// 转换成json
					marshal, err := json.Marshal(batchData)
					if err != nil {
						log.Printf("json.Marshal err: %v", err)
						return
					}

					// 发送 HTTP POST 请求
					_, err = utils.HttpPostRequest(utils.EDB_LIB_URL+utils.ADD_BATCH_RZD_DATA, 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["Source"] = utils.DATA_SOURCE_RZD
				postEdbLib, err := httpRequestFill(paramsLib, utils.GET_RZD_EDB_INFO_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(utils.StringToTimeFormat(index.DataTime, utils.FormatDate).UnixMilli()),
					}
					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
				}
			}

		}
	}
}

func login(ctx context.Context) error {

	return chromedp.Run(ctx,
		chromedp.Navigate(rzdLoginPath),
		chromedp.SetValue(`input[id="Username"]`, utils.RZD_USERNAME, chromedp.ByQuery),
		chromedp.SetValue(`input[id="Password"]`, utils.RZD_PASSWORD, chromedp.ByQuery),
		chromedp.WaitEnabled(`//button[text()='Login']`, chromedp.BySearch),
		chromedp.Click(`//button[text()='Login']`, chromedp.BySearch),
		chromedp.Sleep(5*time.Second),

		// 等待并点击登录后页面的链接
		chromedp.WaitVisible(`a[href="/clients/"]`, chromedp.ByQuery), // 等待 Analytics Library 链接可见
		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
}