rustfs-s3-toolkit/examples/basic_usage.py

#!/usr/bin/env python3
"""
RustFS S3 Storage Toolkit 基本使用示例
演示所有 9 个核心功能的使用方法
已在 RustFS 1.0.0-alpha.34 上完成测试
"""

import tempfile
from pathlib import Path
from datetime import datetime

from rustfs_s3_toolkit import S3StorageToolkit


def main():
    """RustFS S3 Storage Toolkit 使用示例"""

    # 1. 初始化工具包
    print("🚀 RustFS S3 Storage Toolkit 使用示例")
    print("🧪 测试环境: RustFS 1.0.0-alpha.34")
    print("=" * 50)

    # 配置信息 - 请根据实际 RustFS 情况修改
    config = {
        "endpoint_url": "https://your-rustfs-endpoint.com",
        "access_key_id": "your-access-key-id",
        "secret_access_key": "your-secret-access-key",
        "bucket_name": "your-bucket-name",
        "region_name": "us-east-1"
    }
    
    # 创建工具包实例
    toolkit = S3StorageToolkit(**config)
    print("✅ 工具包初始化完成")
    
    # 2. 测试连接
    print("\n📡 测试连接...")
    result = toolkit.test_connection()
    if result['success']:
        print(f"✅ 连接成功: {result['message']}")
        print(f"📊 存储桶数量: {result['bucket_count']}")
        print(f"🎯 目标存储桶存在: {result['target_bucket_exists']}")
    else:
        print(f"❌ 连接失败: {result['error']}")
        return
    
    # 3. 上传文件
    print("\n📤 上传文件...")
    
    # 创建测试文件
    test_content = f"Hello RustFS S3 Storage Toolkit!\n创建时间: {datetime.now().isoformat()}"
    with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt', encoding='utf-8') as f:
        f.write(test_content)
        test_file = f.name
    
    # 上传文件
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    remote_key = f"examples/test_file_{timestamp}.txt"
    
    result = toolkit.upload_file(test_file, remote_key)
    if result['success']:
        print(f"✅ 文件上传成功")
        print(f"📁 远程路径: {result['key']}")
        print(f"🔗 公开链接: {result['public_url']}")
        print(f"📊 文件大小: {result['file_size']} 字节")
    else:
        print(f"❌ 文件上传失败: {result['error']}")
    
    # 4. 创建文件夹
    print("\n📁 创建文件夹...")
    folder_path = f"examples/test_folder_{timestamp}/"
    
    result = toolkit.create_folder(folder_path)
    if result['success']:
        print(f"✅ 文件夹创建成功: {result['folder_path']}")
    else:
        print(f"❌ 文件夹创建失败: {result['error']}")
    
    # 5. 上传目录
    print("\n📂 上传目录...")
    
    # 创建测试目录
    test_dir = tempfile.mkdtemp(prefix='s3_example_')
    test_dir_path = Path(test_dir)
    
    # 创建一些测试文件
    (test_dir_path / "file1.txt").write_text("这是文件1", encoding='utf-8')
    (test_dir_path / "file2.txt").write_text("这是文件2", encoding='utf-8')
    
    # 创建子目录
    sub_dir = test_dir_path / "subdir"
    sub_dir.mkdir()
    (sub_dir / "file3.txt").write_text("这是子目录中的文件", encoding='utf-8')
    
    # 上传目录
    remote_prefix = f"examples/test_directory_{timestamp}/"
    result = toolkit.upload_directory(test_dir, remote_prefix)
    if result['success']:
        print(f"✅ 目录上传成功")
        print(f"📁 本地目录: {result['local_directory']}")
        print(f"🌐 远程前缀: {result['remote_prefix']}")
        print(f"📊 文件数量: {result['file_count']}")
        print("📄 上传的文件:")
        for file_key in result['uploaded_files']:
            print(f"   - {file_key}")
    else:
        print(f"❌ 目录上传失败: {result['error']}")
    
    # 6. 列出文件
    print("\n📋 列出文件...")
    result = toolkit.list_files(prefix="examples/", max_keys=10)
    if result['success']:
        print(f"✅ 文件列表获取成功")
        print(f"📊 文件数量: {result['file_count']}")
        print("📄 文件列表:")
        for file_info in result['files']:
            size_mb = file_info['size'] / (1024 * 1024)
            print(f"   - {file_info['key']} ({size_mb:.3f} MB)")
    else:
        print(f"❌ 文件列表获取失败: {result['error']}")
    
    # 7. 下载文件
    print("\n📥 下载文件...")
    download_path = tempfile.mktemp(suffix='_downloaded.txt')
    
    result = toolkit.download_file(remote_key, download_path)
    if result['success']:
        print(f"✅ 文件下载成功")
        print(f"📁 本地路径: {result['local_path']}")
        print(f"📊 文件大小: {result['file_size']} 字节")
        
        # 验证下载的内容
        with open(download_path, 'r', encoding='utf-8') as f:
            downloaded_content = f.read()
        
        if test_content == downloaded_content:
            print("✅ 文件内容验证成功")
        else:
            print("❌ 文件内容验证失败")
    else:
        print(f"❌ 文件下载失败: {result['error']}")
    
    # 8. 下载目录
    print("\n📂 下载目录...")
    download_dir = tempfile.mkdtemp(prefix='s3_download_')
    
    result = toolkit.download_directory(remote_prefix, download_dir)
    if result['success']:
        print(f"✅ 目录下载成功")
        print(f"📁 本地目录: {result['local_directory']}")
        print(f"🌐 远程前缀: {result['remote_prefix']}")
        print(f"📊 文件数量: {result['file_count']}")
        print("📄 下载的文件:")
        for file_path in result['downloaded_files']:
            print(f"   - {file_path}")
    else:
        print(f"❌ 目录下载失败: {result['error']}")
    
    # 9. 删除文件
    print("\n🗑️ 删除文件...")
    result = toolkit.delete_file(remote_key)
    if result['success']:
        print(f"✅ 文件删除成功: {result['key']}")
    else:
        print(f"❌ 文件删除失败: {result['error']}")
    
    # 10. 删除目录
    print("\n🗑️ 删除目录...")
    result = toolkit.delete_directory(remote_prefix)
    if result['success']:
        print(f"✅ 目录删除成功")
        print(f"📊 删除文件数量: {result['deleted_count']}")
    else:
        print(f"❌ 目录删除失败: {result['error']}")
    
    # 清理本地文件
    import os
    import shutil
    
    try:
        os.unlink(test_file)
        os.unlink(download_path)
        shutil.rmtree(test_dir)
        shutil.rmtree(download_dir)
        print("\n🧹 本地临时文件清理完成")
    except:
        pass
    
    print("\n🎉 示例演示完成！")


if __name__ == "__main__":
    main()