jx_ld_server/utils/myTool.go

package utils

import (
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net"
	"os"
	"path/filepath"
	"server/dao"
	"server/logger"
	"strings"
	"time"
)

const dataFilePath = "static/data.json"

// LoadData 从文件中加载 Region 数据。
func LoadData() ([]dao.Region, error) {
	data, err := os.ReadFile(dataFilePath)

	datafb, err := os.ReadFile("static/data-fb.json")
	var regionsfb []dao.Region
	if err := json.Unmarshal(datafb, &regionsfb); err != nil {
		// 提供更多的上下文信息来帮助调试
		return nil, fmt.Errorf("解析 JSON 失败: %v, 原始数据: %s", err, string(data))
	}

	if err != nil {
		if os.IsNotExist(err) {
			return regionsfb, nil // 文件不存在时返回空数组
		}
		return nil, fmt.Errorf("读取文件失败: %v", err)
	}

	// 检查文件是否为空
	if len(data) == 0 {
		return regionsfb, nil // 如果文件为空，返回空数组
	}

	var regions []dao.Region
	if err := json.Unmarshal(data, &regions); err != nil {
		// 提供更多的上下文信息来帮助调试
		return nil, fmt.Errorf("解析 JSON 失败: %v, 原始数据: %s", err, string(data))
	}

	return regions, nil
}

func GetOnlineDevices() (devices []dao.Device, err error) {
	regions, err := LoadData()
	if err != nil {
		return nil, err
	}

	for _, region := range regions {
		for _, device := range region.Devices {
			if device.State == 0 {
				continue
			}
			devices = append(devices, device)
		}
	}

	return devices, err
}

func GetAllDevices() (devices []dao.Device, err error) {
	regions, err := LoadData()
	if err != nil {
		return nil, err
	}

	for _, region := range regions {
		for _, device := range region.Devices {
			devices = append(devices, device)
		}
	}

	return devices, err
}

// GetDataByDeviceId 按设备id获取信息 地区+设备
func GetDataByDeviceId(deviceId string) (reg dao.Region, dev dao.Device, err error) {
	regions, err := LoadData()
	if err != nil {
		return reg, dev, err
	}

	for _, region := range regions {
		for _, device := range region.Devices {
			if device.Sn == deviceId {
				return region, device, nil
			}
		}
	}

	return reg, dev, fmt.Errorf("设备%s未找到", deviceId)
}

func SaveRegionOnData(data dao.Region) ([]dao.Region, error) {
	regions, err := LoadData()
	if err != nil {
		return regions, err
	}
	for i, region := range regions {
		if region.Name == data.Name {
			regions[i] = data
		}
	}
	return regions, nil
}

// SaveData 保存数据到文件
func SaveData(path string, parameter interface{}) error {
	// 如果 parameter 是 nil 或者是空的结构体/切片，可以选择不写入文件或返回错误。
	if parameter == nil {
		return fmt.Errorf("无法保存 nil 数据")
	}

	// 序列化数据
	data, err := json.MarshalIndent(parameter, "", "  ")
	if err != nil {
		return fmt.Errorf("JSON 序列化失败: %v", err)
	}

	// 创建临时文件用于原子写入
	tempFile, err := ioutil.TempFile(filepath.Dir(path), "tmp-")
	if err != nil {
		return fmt.Errorf("创建临时文件失败: %v", err)
	}
	defer os.Remove(tempFile.Name()) // 确保即使出错也能清理临时文件

	// 写入临时文件
	if _, err := tempFile.Write(data); err != nil {
		tempFile.Close()
		return fmt.Errorf("写入临时文件失败: %v", err)
	}

	// 确保所有数据都写入磁盘
	if err := tempFile.Sync(); err != nil {
		tempFile.Close()
		return fmt.Errorf("同步临时文件失败: %v", err)
	}

	// 关闭临时文件
	if err := tempFile.Close(); err != nil {
		return fmt.Errorf("关闭临时文件失败: %v", err)
	}

	// 使用原子操作替换原始文件
	if err := os.Rename(tempFile.Name(), path); err != nil {
		return fmt.Errorf("替换文件失败: %v", err)
	}

	return nil
}

const (
	maxRetries    = 3               // 最大重试次数
	writeTimeout  = 5 * time.Second // 写入超时时间
	reconnectWait = 2 * time.Second // 重连等待时间
)

func WriteDevice(frame []byte, conn net.Conn) error {
	for attempts := 0; attempts < maxRetries; attempts++ {
		if err := conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil {
			logger.Get().Errorf("Set write deadline failed: %v", err)
			return err
		}

		_, err := conn.Write(frame)
		if err == nil {
			return nil
		}

		// 检查是否是对端强制关闭连接的错误
		if ne, ok := err.(*net.OpError); ok && (strings.Contains(ne.Err.Error(), "forcibly closed") || strings.Contains(ne.Err.Error(), "broken pipe") || strings.Contains(ne.Err.Error(), "connection reset")) {
			logger.Get().Warnf("Connection forcibly closed by peer, retrying (%d/%d)", attempts+1, maxRetries)

			// 关闭旧连接
			if err := conn.Close(); err != nil {
				logger.Get().Errorf("Failed to close connection: %v", err)
			}

			// 尝试重新建立连接
			var newConn net.Conn
			remoteAddr := conn.RemoteAddr().String()

			// 解析原始连接的远程地址和网络接口
			addr, networkInterface, err := parseRemoteAddr(remoteAddr)
			if err != nil {
				logger.Get().Errorf("Failed to parse remote address: %v", err)
				continue // 继续下一次重试
			}

			newConn, err = net.Dial("tcp", fmt.Sprintf("%s%%%s", addr, networkInterface))
			if err != nil {
				logger.Get().Errorf("Reconnect failed: %v", err)
				time.Sleep(reconnectWait) // 等待一段时间后重试
				continue                  // 继续下一次重试
			}
			conn = newConn
			continue // 重试写入
		}

		// 如果不是对端强制关闭连接的错误，则直接返回错误
		logger.Get().Errorf("Write failed: %v", err)
		return fmt.Errorf("write failed after %d retries: %v", attempts+1, err)
	}

	return fmt.Errorf("failed to write after %d retries", maxRetries)
}

// 解析远程地址，提取 IP 地址和网络接口名称
func parseRemoteAddr(addr string) (string, string, error) {
	parts := strings.Split(addr, "%")
	if len(parts) != 2 {
		return "", "", fmt.Errorf("invalid remote address format: %s", addr)
	}
	return parts[0], parts[1], nil
}

func WriteAndReadDevice(frame []byte, conn net.Conn, former, after int) (data []byte, err error) {
	// 发送 Modbus RTU 帧
	n, err := conn.Write(frame)
	if err != nil {
		logger.Get().Errorln("Error writing to connection:", err)
		return
	}

	// 等待一段时间以接收响应
	time.Sleep(1000 * time.Millisecond)

	// 读取响应
	buffer := make([]byte, 1024)
	n, err = conn.Read(buffer)
	if err != nil {
		logger.Get().Errorln("Error reading from connection:", err)
		return
	}

	// 检查读取的字节数是否足够
	if n < former+after {
		err = fmt.Errorf("not enough bytes read to satisfy the slice range")
		return
	}

	// 返回子切片
	return buffer[former : n-after], err
}