LQAdminPlatform/src/test/test_minio.py

import io
from datetime import timedelta
from minio import Minio
from minio.error import S3Error

# 1. 初始化客户端
# 配置信息来源于用户最新提供
ENDPOINT = "192.168.91.15:19000"
ACCESS_KEY = "2I8pipSfCa7tgBzGEZQJ"
SECRET_KEY = "468kSI3wKWix5ThlBOVuRRUst8ImudH3aSCyOwmQ"
BUCKET_NAME = "auto"
BASE_PATH = "sampledata"

client = Minio(
    ENDPOINT,
    access_key=ACCESS_KEY,
    secret_key=SECRET_KEY,
    secure=False  # 使用 HTTP
)

def upload_file(file_path: str, content: bytes, content_type: str = "application/octet-stream"):
    """
    上传文件到 MinIO
    Args:
        file_path: 文件路径（相对于 sampledata/），如 "test/hello.txt"
        content: 文件内容（bytes）
        content_type: MIME 类型
    """
    object_name = f"{BASE_PATH}/{file_path}"
    try:
        result = client.put_object(
            BUCKET_NAME,
            object_name,
            io.BytesIO(content),
            length=len(content),
            content_type=content_type
        )
        print(f"上传成功：{result.object_name}, ETag: {result.etag}")
        return result
    except S3Error as e:
        print(f"上传失败: {e}")

def download_file(file_path: str) -> bytes:
    """
    从 MinIO 下载文件
    Args:
        file_path: 文件路径（相对于 sampledata/）
    Returns:
        文件内容 (bytes)
    """
    object_name = f"{BASE_PATH}/{file_path}"
    try:
        response = client.get_object(BUCKET_NAME, object_name)
        content = response.read()
        response.close()
        response.release_conn()
        return content
    except S3Error as e:
        print(f"下载失败: {e}")
        return b""

def list_files(prefix: str = "", recursive: bool = True):
    """
    列出指定路径下的文件
    Args:
        prefix: 路径前缀 (相对于 sampledata/)
        recursive: 是否递归列出子目录
    """
    full_prefix = f"{BASE_PATH}/{prefix}" if prefix else f"{BASE_PATH}/"
    try:
        objects = client.list_objects(BUCKET_NAME, prefix=full_prefix, recursive=recursive)
        files = []
        for obj in objects:
            files.append({
                "name": obj.object_name,
                "size": obj.size,
                "last_modified": obj.last_modified
            })
        return files
    except S3Error as e:
        print(f"获取文件列表失败: {e}")
        return []

def delete_file(file_path: str):
    """
    删除文件
    Args:
        file_path: 文件路径（相对于 sampledata/）
    """
    object_name = f"{BASE_PATH}/{file_path}"
    try:
        client.remove_object(BUCKET_NAME, object_name)
        print(f"删除成功：{object_name}")
    except S3Error as e:
        print(f"删除失败: {e}")

def get_presigned_url(file_path: str, expires: int = 3600):
    """
    生成预签名下载 URL
    Args:
        file_path: 文件路径（相对于 sampledata/）
        expires: 过期时间（秒），默认1小时
    """
    object_name = f"{BASE_PATH}/{file_path}"
    try:
        url = client.presigned_get_object(
            BUCKET_NAME,
            object_name,
            expires=timedelta(seconds=expires)
        )
        return url
    except S3Error as e:
        print(f"生成预签名URL失败: {e}")

def get_presigned_upload_url(file_path: str, expires: int = 3600):
    """
    生成预签名上传 URL
    Args:
        file_path: 文件路径（相对于 sampledata/）
        expires: 过期时间（秒），默认1小时
    """
    object_name = f"{BASE_PATH}/{file_path}"
    try:
        url = client.presigned_put_object(
            BUCKET_NAME,
            object_name,
            expires=timedelta(seconds=expires)
        )
        return url
    except S3Error as e:
        print(f"生成上传URL失败: {e}")

if __name__ == "__main__":
    # 测试流程
    test_filename = "file/hello_world.txt"
    test_content = b"Hello, MinIO! This is a test file."

    print("--- 1. 上传测试 ---")
    upload_file(test_filename, test_content, "text/plain")

    print("\n--- 2. 列表测试 ---")
    files = list_files("test/")
    for f in files:
        print(f"文件: {f['name']}，大小：{f['size']} bytes")

    print("\n--- 3. 下载测试 ---")
    content = download_file(test_filename)
    print(f"下载内容: {content.decode('utf-8')}")

    print("\n--- 4. 预签名 URL 测试 ---")
    download_url = get_presigned_url(test_filename)
    print(f"下载链接：{download_url}")
    
    upload_url = get_presigned_upload_url("test/upload_via_url.txt")
    print(f"上传链接：{upload_url}")
    print("(可使用 requests.put(upload_url, data=b'content') 测试上传)")

    print("\n--- 5. 删除测试 ---")
    delete_file(test_filename)
    
    print("\n测试完成。")