mysteel_watch/watch/watch.go

package watch

import (
	"fmt"
	"hongze/mysteel_watch/global"
	"hongze/mysteel_watch/models/index"
	"hongze/mysteel_watch/utils"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/fsnotify/fsnotify"
	"github.com/xuri/excelize/v2"
)

func ListenFolderNew() {
	fmt.Println("-----文件夹监听-------")
	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		log.Fatal(err)
	}
	defer watcher.Close()

	done2 := make(chan bool)
	go func() {
		for {
			select {
			case event, ok := <-watcher.Events:
				if ok && event.Op == fsnotify.Create &&
					!strings.Contains(event.Name, "tmp") &&
					!strings.Contains(event.Name, ".TMP") &&
					!strings.Contains(event.Name, "~") &&
					(strings.Contains(event.Name, "xlsx") || strings.Contains(event.Name, "xls")) {
					WatchIndexFile(event.Name)
				}
			case err := <-watcher.Errors:
				log.Println("error:", err)
			case <-time.After(60 * time.Second):
				continue
			}
		}
	}()
	err = watcher.Add(utils.IndexSaveDir)
	if err != nil {
		log.Fatal(err)
	}
	<-done2
}

//检测指标文件
func WatchIndexFile(filePath string) {
	fmt.Println("filePath:", filePath)
	//filePath:D:\mysteel_data\CM0000568866_release.xlsx
	time.Sleep(10 * time.Second)
	if !utils.FileIsExist(filePath) {
		fmt.Println("filePath is not exist:" + filePath)
		return
	}
	//读取文件内容
	global.LOG.Info("WatchFile:" + filePath)
	f, err := excelize.OpenFile(filePath)
	global.LOG.Info("OpenFile:" + filePath)
	if err != nil {
		fmt.Println("OpenFile:" + filePath + ",Err:" + err.Error())
		return
	}
	var newFilePath string
	defer func() {
		if err := f.Close(); err != nil {
			fmt.Println("FileClose Err:" + err.Error())
			return
		}
		//重命名文件
		if filePath != newFilePath {
			err := os.Rename(filePath, newFilePath)
			if err != nil {
				fmt.Println("os.Rename Err:" + err.Error())
			}
		}
	}()
	var runMode string
	if strings.Contains(filePath, "debug") {
		runMode = "debug"
	} else {
		runMode = "release"
	}

	dir, fp := filepath.Split(filePath)

	var wg = sync.WaitGroup{}
	wg.Add(1)
	go func() {
		sheetList := f.GetSheetList()
		for _, sv := range sheetList {

			var indexName, indexCode, unit, source, frequency, startDate, endDate, describe string
			var indexId int64
			rows, err := f.GetRows(sv)
			if err != nil {
				fmt.Println("f.GetRows:err:" + err.Error())
				return
			}

			indexObj := new(index.BaseFromMysteelChemicalIndex)
			dataList := make([]index.BaseFromMysteelChemicalData, 0)

			dataMap := make(map[string]string)
			for rk, row := range rows {
				if rk > 0 {
					if rk < 10 {
						for ck, colCell := range row {
							if ck == 1 {
								if rk == 1 {
									indexName = colCell
								}
								if rk == 2 {
									unit = colCell
								}
								if rk == 3 {
									source = colCell
								}
								if rk == 4 {
									indexCode = colCell
								}
								if rk == 5 {
									frequency = colCell
									if !strings.Contains(frequency, "度") {
										frequency = frequency + "度"
									}
								}
								if rk == 6 {
									dateArr := strings.Split(colCell, "～")
									if len(dateArr) >= 2 {
										startDate = dateArr[0]
										endDate = dateArr[1]
									}
								}
								if rk == 7 {
									describe = colCell
								}
							}
						}

						if rk == 9 {
							if indexName == "" {
								global.LOG.Info("未刷新到指标数据:filePath:" + filePath)
								break
							}
							//判断指标是否存在
							var isAdd int
							item, err := indexObj.GetIndexItem(runMode, indexCode)
							if err != nil {
								if err.Error() == "record not found" {
									isAdd = 1
								} else {
									isAdd = -1
									fmt.Println("GetIndexItem Err:" + err.Error())
									return
								}
							}
							if item != nil && item.BaseFromMysteelChemicalIndexId > 0 {
								fmt.Println("item:", item)
								isAdd = 2
							} else {
								isAdd = 1
							}

							fmt.Println("isAdd:", isAdd)
							if !strings.Contains(frequency, "度") {
								frequency = frequency + "度"
							}

							var frequencyStr string
							if strings.Contains(frequency, "日") {
								frequencyStr = "day"
							} else if strings.Contains(frequency, "周") {
								frequencyStr = "week"
							} else if strings.Contains(frequency, "月") || strings.Contains(frequency, "旬") {
								frequencyStr = "month"
							} else if strings.Contains(frequency, "年") {
								frequencyStr = "year"
							}
							frequencyStr = "_" + frequencyStr
							if !strings.Contains(filePath, frequencyStr) {
								fpArr := strings.Split(fp, "_")
								for k, v := range fpArr {
									if k == 0 {
										newFilePath = v + frequencyStr
									} else {
										newFilePath = newFilePath + "_" + v
									}
								}
								newFilePath = dir + newFilePath
							} else {
								newFilePath = filePath
							}

							if isAdd == 1 {
								indexObj.IndexCode = indexCode
								indexObj.IndexName = indexName
								indexObj.Unit = unit
								indexObj.Source = source
								indexObj.Describe = describe
								indexObj.StartDate, _ = time.ParseInLocation(utils.FormatDate, startDate, time.Local)
								indexObj.EndDate, _ = time.ParseInLocation(utils.FormatDate, endDate, time.Local)
								indexObj.Frequency = frequency
								indexObj.FilePath = newFilePath
								err = indexObj.Add(runMode)
								if err != nil {
									fmt.Println("add err:" + err.Error())
									return
								}
								indexId = indexObj.BaseFromMysteelChemicalIndexId
							} else if isAdd == 2 {
								indexObj.IndexCode = indexCode
								indexObj.IndexName = indexName
								indexObj.Unit = unit
								indexObj.Source = source
								indexObj.Describe = describe
								indexObj.StartDate, _ = time.ParseInLocation(utils.FormatDate, startDate, time.Local)
								indexObj.EndDate, _ = time.ParseInLocation(utils.FormatDate, endDate, time.Local)
								indexObj.Frequency = frequency
								indexObj.FilePath = newFilePath
								indexObj.ModifyTime = time.Now()
								indexId = item.BaseFromMysteelChemicalIndexId
								//修改数据
								updateColsArr := make([]string, 0)
								updateColsArr = append(updateColsArr, "index_name")
								updateColsArr = append(updateColsArr, "unit")
								updateColsArr = append(updateColsArr, "source")
								updateColsArr = append(updateColsArr, "frequency")
								updateColsArr = append(updateColsArr, "start_date")
								updateColsArr = append(updateColsArr, "end_date")
								updateColsArr = append(updateColsArr, "describe")
								updateColsArr = append(updateColsArr, "end_date")
								updateColsArr = append(updateColsArr, "modify_time")
								updateColsArr = append(updateColsArr, "file_path")

								indexObj.Update(runMode, updateColsArr)

								dataObj := new(index.BaseFromMysteelChemicalData)
								//获取已存在的所有数据
								dataList, err := dataObj.GetIndexDataList(runMode, indexCode)
								if err != nil {
									fmt.Println("GetIndexDataList Err:" + err.Error())
									return
								}
								fmt.Println("dataListLen:", len(dataList))
								for _, v := range dataList {
									dateStr := v.DataTime.Format(utils.FormatDate)
									dataMap[dateStr] = v.Value
								}
							}
						}
					} else {
						var date, value string
						for ck, colCell := range row {
							if ck == 0 {
								date = colCell
							} else {
								value = colCell
							}
						}
						if _, ok := dataMap[date]; !ok {
							dateTime, err := time.ParseInLocation(utils.FormatDate, date, time.Local)
							if err != nil {
								fmt.Println("time.ParseInLocation Err:" + err.Error())
								return
							}
							dataItem := new(index.BaseFromMysteelChemicalData)
							dataItem.BaseFromMysteelChemicalIndexId = indexId
							dataItem.IndexCode = indexCode
							dataItem.DataTime = dateTime
							dataItem.Value = value
							dataItem.CreateTime = time.Now()
							dataItem.ModifyTime = time.Now()
							dataList = append(dataList, *dataItem)
						}
					}
				}
			}

			if len(dataList) > 0 {
				dataObj := new(index.BaseFromMysteelChemicalData)
				err = dataObj.Add(runMode, dataList)
				if err != nil {
					fmt.Println("dataObj.Add() Err:" + err.Error())
				}
			}
		}
		wg.Done()
	}()
	wg.Wait()
}

/*
CREATE动作即临时文件的创建
WRITE写文件动作
CHMOD修改文件属性
REMOVE删除临时文件。
*/