jwebfs/scagent/core/FileRpc.go

package core

import (
	"bufio"
	"fmt"
	"io"
	"net/rpc"
	"os"
	"path"
	"path/filepath"
	"scagent/config"
	"scagent/util"
	"strings"
	"time"

	"go.uber.org/zap"
)

// 压缩文件
type ZipBlock struct {
	Basepath   string
	Curpath    string   // 当前路径
	SelFile    []string // 选中的文件
	RemoteAddr string   // 远程主机地址，带端口
	RemotePath string   // 远程主机的存储路径
}

// FileTransferService 定义文件传输服务
type FileService struct{}

// FileInfo 包含文件的元信息
type FileInfo struct {
	FileName string
	FileSize int64
}

// FileChunk 表示文件的一部分
type FileChunk struct {
	Data     []byte
	FileName string
	Offset   int64
	IsLast   bool
}

// SendFile 接收文件信息（名称和大小）
func (f *FileService) SendFileInfo(info FileInfo, reply *bool) error {
	// 创建保存文件的目录
	// if err := os.MkdirAll("received_files", 0755); err != nil {
	// 	return err
	// }

	// 提取路径 /www/wwwroot/bidemo.com/BIU_20250918_183144.zip
	dirPath := filepath.Dir(info.FileName)
	// 创建目录
	if err := os.MkdirAll(dirPath, 0755); err != nil {
		return err
	}

	// 创建文件（如果存在则截断）
	filePath := filepath.Join("received_files", info.FileName)
	file, err := os.Create(filePath)
	if err != nil {
		return err
	}
	defer file.Close()

	*reply = true
	return nil
}

// SendFileChunk 接收文件块并写入文件
func (f *FileService) SendFileChunk(chunk FileChunk, reply *bool) error {
	// filePath := filepath.Join("received_files", chunk.FileName)
	fmt.Printf("recive file :%s", chunk.FileName)
	// 合并为实际路径
	// filePath := filepath.Join("received_files", chunk.FileName)
	filePath := chunk.FileName

	// 打开文件，使用追加模式
	file, err := os.OpenFile(filePath, os.O_WRONLY, 0644)
	if err != nil {
		return err
	}
	defer file.Close()

	// 移动到指定偏移量
	if _, err := file.Seek(chunk.Offset, io.SeekStart); err != nil {
		return err
	}

	// 写入数据
	if _, err := file.Write(chunk.Data); err != nil {
		return err
	}

	*reply = true
	return nil
}

// 压缩文件
// 返回压缩包的名称
func (f *FileService) Compress(args *ZipBlock, reply *string) error {
	// fmt.Printf("compress args: %v\n", args)
	// 启用日志
	logger := util.NewProductionLogger()
	defer logger.Sync()
	// 实际路径
	realFilePath := filepath.Join(args.Basepath, args.Curpath)
	logger.Info("Compress", zap.String("realFilePath", realFilePath))
	// zip 文件名
	zpFileName := "BIU_" + time.Now().Format("20060102_150405") + ".zip"
	// 创建zip 异步？
	taskId := make(chan string)
	// 异步压缩
	go func() {
		util.CompressToZip(zpFileName, realFilePath, args.SelFile)
		taskId <- "arcok"
		// 日志输出
		logger.Info("压缩文件", zap.String("filename", zpFileName), zap.String("path", realFilePath))
	}()
	// ZIP 文件的实际路径
	ziprl := path.Join(config.G.FilePath, "/sync_zips", zpFileName)
	// zip 创建成功后
	rest := <-taskId
	// 如果压缩包生成成功以后再执行
	if strings.EqualFold(strings.ToLower(rest), "arcok") {
		// 合并为实际路径
		remotePath := path.Join("/www/wwwroot", args.RemotePath, zpFileName)
		// 异步发送文件到远程服务器上
		go clientUploadFile(args.RemoteAddr, ziprl, remotePath)

		// 返回压缩包名称
		*reply = ziprl
	}

	return nil
}

// 客户端调用RPC传送文件的函数
// 传入zip的实际路径，远程的服务器带端口，远程文件存放路径
func clientUploadFile(remote string, filePath string, uploadPath string) error {
	// fmt.Printf("remote: %s, filePath: %s, uploadPath: %s\n", remote, filePath, uploadPath)
	// 启用日志
	logger := util.NewProductionLogger()
	defer logger.Sync()
	// 将服务器的信息拆分为ip和端口
	harr := strings.Split(remote, ":")
	dstip := harr[0]
	dport := harr[1]

	// 连接到RPC服务器
	service := fmt.Sprintf("%v:%v", dstip, dport)
	client, err := rpc.Dial("tcp", service)
	if err != nil {
		//panic(err)
		fmt.Printf("连接到RPC服务器失败: %v", err)
		return err
	}
	defer client.Close()
	// 获取文件信息
	fileInfo, err := os.Stat(filePath)
	if err != nil {
		panic(err)
	}

	// 提取文件名
	fileName := filepath.Base(filePath)
	// 发送文件信息
	var reply bool
	err = client.Call("FileService.SendFileInfo", FileInfo{
		FileName: fileName,
		FileSize: fileInfo.Size(),
	}, &reply)
	if err != nil || !reply {
		logger.Error("发送文件信息失败", zap.Error(err))
	}
	// 打开文件准备读取
	file, err := os.Open(filePath)
	if err != nil {
		panic(err)
	}
	defer file.Close()

	// 读取并发送文件块
	chunkSize := 1 * 1024 * 1024      // 1MB
	buffer := make([]byte, chunkSize) // 4KB块大小
	reader := bufio.NewReader(file)
	var offset int64 = 0

	// logger.Info("开始发送文件: %s (大小: %d bytes)", zap.String("filename", fileName), zap.String("path", filePath))
	fmt.Printf("开始发送文件: %s (大小: %d bytes)\n", fileName, fileInfo.Size())

	// 发送文件
	for {
		n, err := reader.Read(buffer)
		if err != nil {
			break
		}

		// 判断是否是最后一块
		isLast := offset+int64(n) >= fileInfo.Size()

		// 发送文件块
		err = client.Call("FileService.SendFileChunk", FileChunk{
			Data:     buffer[:n],
			FileName: uploadPath,
			Offset:   offset,
			IsLast:   isLast,
		}, &reply)

		if err != nil || !reply {
			logger.Error("发送文件块失败", zap.Error(err))
		}

		offset += int64(n)

		// 显示进度
		progress := float64(offset) / float64(fileInfo.Size()) * 100
		fmt.Printf("\r发送进度: %.2f%%", progress)
	}
	fmt.Println("\n文件发送完成!")

	return nil
}