yx-dataset-server/app/bll/impl/internal/b_file.go

package internal

import (
	"bytes"
	"context"
	"filesdk"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"strings"

	"github.com/gogf/gf/v2/os/glog"
	"yx-dataset-server/app/schema"
	"yx-dataset-server/library/logger"
	"yx-dataset-server/library/minio"
	"yx-dataset-server/library/utils"
)

// NewFile 创建文件管理实例
func NewFile() *File {
	return &File{}
}

// File 文件管理
type File struct{}

// Upload 上传文件
func (a *File) Upload(ctx context.Context, r *http.Request, formKey, bucket string) (*schema.FileInfo, error) {
	file, header, err := r.FormFile(formKey)
	if err != nil {
		return nil, err
	}
	buff := new(bytes.Buffer)
	_, err = io.Copy(buff, file)
	if err != nil {
		return nil, err
	}
	headers := map[string]string{}
	headers["FILE-EXPIRE"] = utils.GetConfig("file_server.file-expire").String()
	result, err := filesdk.GetHandle().Upload(header.Filename, bucket, headers, buff.Bytes())
	if err != nil {
		glog.Debugf(ctx, "上传文件失败:%s", err.Error())
		return nil, err
	}
	err = filesdk.GetHandle().HandlePersistent(result.Hash)
	if err != nil {
		logger.Printf(ctx, "文件持久化设置失败：%s", err.Error())
	}

	info := &schema.FileInfo{
		URL:  result.URL,
		Name: result.Name,
		Size: result.Size,
		MD5:  result.Hash,
	}

	return info, nil
}

// Download 下载文件
func (a *File) Download(ctx context.Context, fileUrl string) ([]byte, string, error) {
	Url, err := url.PathUnescape(fileUrl)
	if err != nil {
		return nil, "", err
	}
	stat, err := minio.GetClient().Stat(Url)
	if err != nil {
		fmt.Println("文件不存在")
		return nil, "", err
	}
	fileData, err := a.getSourceFileData(ctx, Url)
	if err != nil {
		return nil, "", err
	}

	return fileData, stat.ContentType, nil

}

func (a *File) getSourceFileData(ctx context.Context, file string) ([]byte, error) {
	var fileData []byte
	obj, err := minio.GetClient().Get(ctx, file)
	if err != nil {
		return nil, err
	}
	buf := new(bytes.Buffer)
	_, _ = io.Copy(buf, obj)

	_ = obj.Close()
	fileData = buf.Bytes()
	return fileData, nil
}

// 修正文件名，将半角 % 替换为全角 ％（不替换的话文件将无法从浏览器中打开）
func (a *File) getFileName(fileName string) string {
	return strings.ReplaceAll(fileName, "%", "％")
}

func (a *File) UploadFromTrainFolder(ctx context.Context, fileName, bucket string) (*schema.FileInfo, error) {
	// 1. 固定目标文件夹路径（和你的目录结构完全匹配）
	// 绝对路径兜底，避免程序运行目录错乱
	baseDir := "/opt/go/yx-dataset-server/file/产品知识"
	// 安全拼接完整路径，防止路径穿越漏洞
	fullFilePath := filepath.Join(baseDir, fileName)

	// 2. 打开本地文件（替代原 r.FormFile 表单取文件逻辑）
	localFile, err := os.Open(fullFilePath)
	if err != nil {
		return nil, fmt.Errorf("读取产品知识文件失败: %w", err)
	}
	// 函数结束自动关闭文件，防止句柄泄漏
	defer localFile.Close()

	// 3. 读取文件到缓冲区（完全保留你原有的内存拷贝逻辑）
	buff := new(bytes.Buffer)
	_, err = io.Copy(buff, localFile)
	if err != nil {
		return nil, fmt.Errorf("文件读取缓存失败: %w", err)
	}

	// 4. 原有SDK上传逻辑1:1保留
	headers := map[string]string{}
	headers["FILE-EXPIRE"] = utils.GetConfig("file_server.file-expire").String()

	result, err := filesdk.GetHandle().Upload(fileName, bucket, headers, buff.Bytes())
	if err != nil {
		glog.Debugf(ctx, "上传文件失败:%s", err.Error())
		return nil, err
	}

	// 5. 文件持久化逻辑完整保留
	err = filesdk.GetHandle().HandlePersistent(result.Hash)
	if err != nil {
		logger.Printf(ctx, "文件持久化设置失败：%s", err.Error())
	}

	// 6. 返回结构完全兼容原定义
	info := &schema.FileInfo{
		URL:  result.URL,
		Name: result.Name,
		Size: result.Size,
		MD5:  result.Hash,
	}

	return info, nil
}

func (a *File) UploadAllTrainFiles(ctx context.Context, path, bucket string) ([]*schema.FileInfo, error) {
	baseDir := fmt.Sprintf("/opt/go/yx-dataset-server/file/%s", path)

	fmt.Printf(fmt.Sprintf("正在读取【%s】目录下的所有文件...\n", path))
	// 读取目录内所有文件
	dirEntries, err := os.ReadDir(baseDir)
	if err != nil {
		return nil, fmt.Errorf("读取文件失败: %w", err)
	}

	var uploadResult []*schema.FileInfo
	fmt.Printf("开始上传文件...\n")
	for k, entry := range dirEntries {
		if entry.IsDir() {
			continue
		}
		fmt.Printf("正在处理第%d个文件：%s\n", k, entry.Name())
		// 调用上面的单文件上传方法
		fileInfo, err := a.UploadFromTrainFolder(ctx, entry.Name(), bucket)
		if err != nil {
			glog.Debugf(ctx, "文件【%s】上传跳过: %s", entry.Name(), err.Error())
			continue
		}

		uploadResult = append(uploadResult, fileInfo)
	}

	return uploadResult, nil
}

// UploadFromLocal 从本地文件路径上传文件
func (a *File) UploadFromLocal(ctx context.Context, filePath string, bucket string) (*schema.FileInfo, error) {
	logger.Printf(ctx, "[文件上传] 开始上传本地文件: %s, bucket: %s", filePath, bucket)

	file, err := os.Open(filePath)
	if err != nil {
		logger.Printf(ctx, "[文件上传] 打开文件失败: %s, 错误: %s", filePath, err.Error())
		return nil, fmt.Errorf("打开文件失败: %s", err.Error())
	}
	defer file.Close()

	// 获取文件信息
	fileInfo, err := file.Stat()
	if err != nil {
		logger.Printf(ctx, "[文件上传] 获取文件信息失败: %s, 错误: %s", filePath, err.Error())
		return nil, fmt.Errorf("获取文件信息失败: %s", err.Error())
	}

	fileSize := fileInfo.Size()
	fileName := fileInfo.Name()
	logger.Printf(ctx, "[文件上传] 文件信息 - 名称: %s, 大小: %d bytes", fileName, fileSize)

	// 读取文件内容
	data, err := io.ReadAll(file)
	if err != nil {
		logger.Printf(ctx, "[文件上传] 读取文件内容失败: %s, 错误: %s", filePath, err.Error())
		return nil, fmt.Errorf("读取文件内容失败: %s", err.Error())
	}

	logger.Printf(ctx, "[文件上传] 文件读取完成: %s, 实际读取: %d bytes", filePath, len(data))

	headers := map[string]string{}
	headers["FILE-EXPIRE"] = utils.GetConfig("file_server.file-expire").String()

	logger.Printf(ctx, "[文件上传] 开始上传到Minio: %s", filePath)
	result, err := filesdk.GetHandle().Upload(fileName, bucket, headers, data)
	if err != nil {
		logger.Printf(ctx, "[文件上传] Minio上传失败: %s, 错误: %s", filePath, err.Error())
		glog.Debugf(ctx, "上传文件失败: %s", err.Error())
		return nil, err
	}

	logger.Printf(ctx, "[文件上传] Minio上传成功: %s -> %s, MD5: %s", filePath, result.URL, result.Hash)

	err = filesdk.GetHandle().HandlePersistent(result.Hash)
	if err != nil {
		logger.Printf(ctx, "[文件上传] 文件持久化设置失败: %s, 错误: %s", filePath, err.Error())
	}

	info := &schema.FileInfo{
		URL:  result.URL,
		Name: result.Name,
		Size: result.Size,
		MD5:  result.Hash,
	}

	logger.Printf(ctx, "[文件上传] 上传完成: %s", filePath)
	return info, nil
}