eta_data_analysis/services/ruizide/data_processor.go

package ruizide

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

	"github.com/chromedp/chromedp"
)

// 定义选择器
var (
	excelDir      = utils.RZD_EXCEL_PATH
	downloadDir   = utils.RZD_DOWNLOAD_PATH
	rzdLoginPath  = utils.RZD_LOGIN_PATH
	rzdBatchSize  = 500
	tableNameList = []string{
		"Oil_Demand_Signals_Weekly_Report",
		"Oil_Supply_Analysis",
		"Supply_Revision_Analysis",
		"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",
		"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",
	}
)

// 处理数据下载的步骤
func downloadData(ctx context.Context) error {
	// Analytics Library
	if err := chromedp.Run(ctx,
		chromedp.Sleep(5*time.Second), // 考虑移除这一行，如果不必要的话
		chromedp.Navigate(rzdLoginPath),
		chromedp.WaitVisible(`a.mat-tooltip-trigger.analytics.home__link-card-icon[href="/analytics"]`, chromedp.ByQuery), // 等待元素可见
		chromedp.Click(`a.mat-tooltip-trigger.analytics.home__link-card-icon[href="/analytics"]`, chromedp.ByQuery),       // 点击链接
		chromedp.Sleep(5*time.Second),
	); err != nil {
		return fmt.Errorf("下载 Analytics Library 数据错误: %v", err)
	}
	if err := chromedp.Run(ctx,
		chromedp.WaitVisible(`#intro-home-page-step1`, chromedp.ByID),                                  // 等待 input 元素可见
		chromedp.SetValue(`#intro-home-page-step1`, "oil demand signals weekly report", chromedp.ByID), // 设置值
		chromedp.SendKeys(`#intro-home-page-step1`, "\u000D", chromedp.ByID),                           // 模拟按下回车键 (\u000D 是回车的 Unicode)
		chromedp.Sleep(5*time.Second),
	); err != nil {
		return fmt.Errorf("设置 input 标签值时发生错误: %v", err)
	}

	if err := chromedp.Run(ctx,
		// 等待第一个 download-btns div 和下载按钮可见
		chromedp.WaitVisible(`//div[contains(@class, 'download-btns')][1]//ul//li//button[contains(@class, 'mat-tooltip-trigger') and contains(@class, 'download__excel') and normalize-space(text())='Data']`, chromedp.BySearch),
		// 点击第一个 div 中的 Excel 下载按钮
		chromedp.Click(`//div[contains(@class, 'download-btns')][1]//ul//li//button[contains(@class, 'mat-tooltip-trigger') and contains(@class, 'download__excel') and normalize-space(text())='Data']`, chromedp.BySearch),
		// 可选：等待下载完成，可以调整等待时间
		chromedp.Sleep(5*time.Second), // 根据下载时间设置
	); err != nil {
		return fmt.Errorf("点击第一个下载按钮时发生错误: %v", err)
	}

	// 解析文件移动到目标目录
	if err := WaitAndRenameDownloadedFile("Oil_Demand_Signals_Weekly_Report_"+utils.GetCurrentYearMonth()+".xlsx", excelDir); err != nil {
		return err
	}

	return nil
}

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

	// 查找下载目录中的文件
	files, err := filepath.Glob(filepath.Join(downloadDir, "*.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)
		}
		// 打印重命名后的文件名
		utils.FileLog.Info("文件重命名并移动到: ", targetFilePath)
		fmt.Printf("文件重命名并移动到: %s\n", targetFilePath)
	}

	return nil
}

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

	dstFile, err := os.Create(destination)
	if err != nil {
		return fmt.Errorf("创建目标文件时出错: %v", err)
	}
	defer func() {
		_ = 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 createPath(paths []string) (err error) {
	for _, path := range paths {
		if path == "" {
			continue
		}
		_, err = os.Stat(path)
		if err != nil {
			if os.IsNotExist(err) {
				if err = os.MkdirAll(path, os.ModePerm); err != nil {
					fmt.Printf("睿咨得创建目录时出错: %v\n", err)
					utils.FileLog.Error("睿咨得创建目录时出错: %v", err)
					return
				}
			} else {
				return
			}
		}

	}
	return
}

// ResolverNet 解析网页数据，下载文件
func ResolverNet(_ context.Context) (err error) {
	if err = createPath([]string{excelDir}); err != nil {
		return
	}
	// 创建 chromedp 执行上下文
	options := append(chromedp.DefaultExecAllocatorOptions[:3],
		//关闭无头模式
		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("safebrowsing.enabled", true), // 启用安全浏览
	)
	parentCtx, cancel := chromedp.NewExecAllocator(context.Background(), options...)

	ctx, ctxCancel := chromedp.NewContext(parentCtx)
	defer func() {
		cancel()
		ctxCancel()
	}()

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

	// 登录操作
	if err = login(ctx); err != nil {
		utils.FileLog.Error("睿咨得登录错误:", err.Error())
		fmt.Printf("睿咨得登录错误: %v\n", err)
		return
	}
	fmt.Printf("登录成功")
	// 下载数据
	if err = downloadData(ctx); err != nil {
		utils.FileLog.Error("睿咨得数据下载错误:", err.Error())
		fmt.Printf("睿咨得数据下载错误: %v\n", err)
		return
	}
	utils.FileLog.Info("睿咨得数据下载完成")
	// 解析表格 读取数据
	FileResolverV2()
	return
}

// 解析本地文件
func FileResolver() error {
	for _, tableName := range tableNameList {
		var fileName string
		// 解析表格
		fileName = tableName + "_" + utils.GetCurrentYearMonth() + ".xlsx"
		filePath := filepath.Join(excelDir, fileName)
		if _, err := os.Stat(filePath); os.IsNotExist(err) {
			utils.FileLog.Error("文件不存在: %v", err)
			continue
		}
		// 打开 Excel 文件
		f, err := excelize.OpenFile(filePath)
		if err != nil {
			utils.FileLog.Error("无法打开 Excel 文件: %v", err)
			return err
		}
		// 获取所有工作表
		sheetNames := f.GetSheetList()

		for _, sheetName := range sheetNames {
			if tableName == "Oil_Demand_Signals_Weekly_Report" && sheetName == "Content" {
				utils.FileLog.Info(tableName+"跳过工作表:", sheetName)
				continue
			}
			utils.FileLog.Info("读取工作表: %s\n", sheetName)
			// 获取工作表的最大行数
			maxRow, err := f.GetRows(sheetName) // 直接获取所有行数据
			if err != nil {
				utils.FileLog.Error("获取工作表数据时出错: %v", err)
				continue
			}
			// 遍历行并打印内容
			var indexData []models.BaseFromRzdData
			for rowIndex, rowData := range maxRow {
				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 {
					utils.FileLog.Error("无法获取处理器: %v", err)
					continue
				}
				baseFromRzdDataList, err := processor.Process(tableName, sheetName, rowIndex, rowData)
				if err != nil {
					utils.FileLog.Error("处理数据失败: %v", err)
					return err
				}
				fmt.Printf("处理数据成功: %v\n", baseFromRzdDataList)
				indexData = append(indexData, baseFromRzdDataList...)
			}
			fmt.Printf("读取工作表完成: %s\n", sheetName)
			utils.FileLog.Info("读取工作表完成:", sheetName)
			// 新增数据源指标数据
			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 {
						utils.FileLog.Error("json.Marshal err: %v", err)
						return err
					}

					// 发送 HTTP POST 请求
					_, err = utils.HttpPostRequest(utils.EDB_LIB_URL+utils.ADD_BATCH_RZD_DATA, string(marshal), "application/json")
					if err != nil {
						utils.FileLog.Error("postEdbLib err: %v", err)
						return err
					}
				}
			}

			// 新增指标库数据
			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 {
					// 有错误就不继续执行
					utils.FileLog.Error("postEdbLib err: %v", err)
					continue
				}
				var requestResponse models.RequestResponse[models.EdbInfo]
				err = json.Unmarshal(postEdbLib, &requestResponse)
				if err != nil {
					utils.FileLog.Error("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 {
					utils.FileLog.Error("postEdbLib err: %v", err)
					return err
				}
				_, err = utils.HttpPostRequest(utils.EDB_LIB_URL+utils.ADD_BATCH_RZD_EDB_DATA, string(marshal), "application/json")
				if err != nil {
					// 有错误就不继续执行
					utils.FileLog.Error("postEdbLib err: %v", err)
					return err
				}
			}
		}
	}
	return nil
}

func login(ctx context.Context) error {
	if rzdLoginPath == "" {
		return errors.New("睿咨得登录页面地址未配置")
	}
	return chromedp.Run(ctx,
		chromedp.Navigate(rzdLoginPath),
		chromedp.WaitVisible(`body`, chromedp.ByQuery),
		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.Sleep(5*time.Second),
		chromedp.Click(`//button[text()='Login']`, chromedp.BySearch),

		// 等待并点击登录后页面的链接
		/*chromedp.WaitVisible(`a[href="/home"]`, 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
}

// 解析本地文件
func FileResolverV2() error {
	//获取rzd数据库分类数据
	rzdClassifyList, err := getRzdClassifyList()
	if err != nil {
		fmt.Printf("获取睿咨得分类数据失败: %v", err)
		utils.FileLog.Error("获取睿咨得分类数据失败:", err.Error())
		return err
	}
	RzdClassifyMap = make(map[string]*models.BaseFromRzdClassify)
	//更新睿咨得分类Map
	for _, classify := range rzdClassifyList {
		RzdClassifyMap[classify.ClassifyName] = classify
	}
	for _, tableName := range tableNameList {
		rzdProcessor, processorErr := GetRZDProcessor(tableName)
		if processorErr != nil {
			utils.FileLog.Error("获取睿咨得数据处理器:", processorErr.Error())
			continue
		}
		err = rzdProcessor.Process(tableName)
		if err != nil {
			utils.FileLog.Error(fmt.Sprintf("%s处理数据失败:", tableName), err.Error())
			continue
		}
	}
	return nil
}