#MinIO 实战教程

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

# 初始化MinIO客户端
def init_minio_client():
    client = Minio(
        "localhost:9000",
        access_key="minioadmin",
        secret_key="minioadmin123",
        secure=False  # 生产环境应设为True
    )
    return client

# 检查存储桶是否存在
def bucket_exists(client, bucket_name):
    return client.bucket_exists(bucket_name)

# 创建存储桶
def create_bucket(client, bucket_name):
    if not client.bucket_exists(bucket_name):
        client.make_bucket(bucket_name)
        print(f"存储桶 {bucket_name} 创建成功")
    else:
        print(f"存储桶 {bucket_name} 已存在")

# 上传文件
def upload_file(client, bucket_name, object_name, file_path):
    client.fput_object(
        bucket_name,
        object_name,
        file_path,
        content_type="application/octet-stream"
    )
    print(f"文件 {file_path} 上传成功至 {bucket_name}/{object_name}")

# 下载文件
def download_file(client, bucket_name, object_name, file_path):
    client.fget_object(bucket_name, object_name, file_path)
    print(f"文件 {bucket_name}/{object_name} 下载成功至 {file_path}")

# 上传对象（从内存）
def upload_object(client, bucket_name, object_name, data):
    client.put_object(
        bucket_name,
        object_name,
        data,
        length=len(data),
        content_type="text/plain"
    )

# 获取对象
def get_object(client, bucket_name, object_name):
    try:
        response = client.get_object(bucket_name, object_name)
        data = response.read()
        response.close()
        response.release_conn()
        return data
    except S3Error as e:
        print(f"获取对象失败: {e}")
        return None

# 列出对象
def list_objects(client, bucket_name, prefix=""):
    objects = client.list_objects(bucket_name, prefix=prefix, recursive=True)
    for obj in objects:
        print(f"Object: {obj.object_name}, Size: {obj.size}")

# 删除对象
def delete_object(client, bucket_name, object_name):
    client.remove_object(bucket_name, object_name)
    print(f"对象 {bucket_name}/{object_name} 删除成功")

# 生成预签名URL
def presigned_get_url(client, bucket_name, object_name, expires=timedelta(hours=1)):
    return client.presigned_get_object(
        bucket_name,
        object_name,
        expires=expires
    )

def presigned_put_url(client, bucket_name, object_name, expires=timedelta(hours=1)):
    return client.presigned_put_object(
        bucket_name,
        object_name,
        expires=expires
    )

# 设置对象生命周期
def set_lifecycle_config(client, bucket_name):
    from minio.commonconfig import ENABLED
    from minio.lifecycleconfig import LifecycleConfig, Rule, Expiration
    from datetime import datetime
    
    config = LifecycleConfig([
        Rule(
            ENABLED,
            rule_filter=LifecycleRuleFilter(prefix="logs/"),
            rule_id="expire-logs-after-30-days",
            expiration=Expiration(days=30)
        )
    ])
    client.set_bucket_lifecycle(bucket_name, config)

# 设置存储桶通知
def set_bucket_notification(client, bucket_name):
    from minio.notificationconfig import NotificationConfig
    from minio.event import ObjectCreatedPut
    
    notification_config = NotificationConfig(
        queue_config_list=[
            QueueConfig(
                "QUEUE-ARN-OF-REDIS",
                [ObjectCreatedPut()],
                config_id="1",
                prefix_filter_rule=PrefixFilterRule("photos/"),
            ),
        ],
    )
    client.set_bucket_notification(bucket_name, notification_config)

# 获取存储桶统计信息
def get_bucket_stats(client, bucket_name):
    # 获取存储桶策略
    policy = client.get_bucket_policy(bucket_name)
    print(f"Bucket Policy: {policy}")
    
    # 获取存储桶配置
    versioning = client.get_bucket_versioning(bucket_name)
    print(f"Versioning Status: {versioning.status}")
    
    # 获取生命周期配置
    lifecycle = client.get_bucket_lifecycle(bucket_name)
    print(f"Lifecycle Config: {lifecycle}")

from flask import Flask, request, jsonify
import uuid
from werkzeug.utils import secure_filename
import mimetypes
import os

app = Flask(__name__)
minio_client = init_minio_client()

# 确保存储桶存在
create_bucket(minio_client, "uploads")

ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif', 'mp4', 'mov'}

def allowed_file(filename):
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

@app.route('/upload', methods=['POST'])
def upload_file_api():
    try:
        if 'file' not in request.files:
            return jsonify({'error': 'No file part'}), 400
        
        file = request.files['file']
        if file.filename == '':
            return jsonify({'error': 'No selected file'}), 400
        
        if file and allowed_file(file.filename):
            filename = secure_filename(file.filename)
            # 使用UUID生成唯一文件名
            unique_filename = f"{uuid.uuid4()}_{filename}"
            bucket_name = "uploads"
            
            # 上传到MinIO
            file_content = file.read()
            minio_client.put_object(
                bucket_name,
                unique_filename,
                file_content,
                length=len(file_content),
                content_type=file.content_type or mimetypes.guess_type(filename)[0] or 'application/octet-stream'
            )
            
            # 生成访问URL
            file_url = f"http://localhost:9000/{bucket_name}/{unique_filename}"
            
            return jsonify({
                'success': True,
                'filename': unique_filename,
                'original_filename': filename,
                'url': file_url,
                'size': len(file_content)
            })
        
        return jsonify({'error': 'File type not allowed'}), 400
    
    except S3Error as e:
        return jsonify({'error': f'MinIO Error: {str(e)}'}), 500
    except Exception as e:
        return jsonify({'error': f'Server Error: {str(e)}'}), 500

@app.route('/download/<filename>')
def download_file_api(filename):
    try:
        bucket_name = "uploads"
        
        # 生成预签名URL
        presigned_url = minio_client.presigned_get_object(
            bucket_name,
            filename,
            expires=timedelta(minutes=10)  # 10分钟有效期
        )
        
        return jsonify({
            'download_url': presigned_url
        })
    
    except S3Error as e:
        return jsonify({'error': f'MinIO Error: {str(e)}'}), 500

if __name__ == '__main__':
    app.run(debug=True)

from PIL import Image
import io
import base64
from typing import Tuple

class ImageStorageService:
    def __init__(self, minio_client, bucket_name="images"):
        self.client = minio_client
        self.bucket_name = bucket_name
        self.allowed_formats = {'JPEG', 'PNG', 'GIF', 'WEBP'}
        
        # 确保存储桶存在
        if not self.client.bucket_exists(self.bucket_name):
            self.client.make_bucket(self.bucket_name)
    
    def upload_image(self, image_data: bytes, filename: str, resize: Tuple[int, int] = None):
        """上传并可选地调整图片大小"""
        # 打开并验证图片
        image_stream = io.BytesIO(image_data)
        image = Image.open(image_stream)
        
        if image.format not in self.allowed_formats:
            raise ValueError(f"Unsupported image format: {image.format}")
        
        # 调整大小（如果指定）
        if resize:
            image = image.resize(resize, Image.Resampling.LANCZOS)
        
        # 保存到字节流
        output_stream = io.BytesIO()
        image.save(output_stream, format=image.format)
        output_stream.seek(0)
        
        # 上传到MinIO
        self.client.put_object(
            self.bucket_name,
            filename,
            output_stream,
            length=output_stream.getbuffer().nbytes,
            content_type=f"image/{image.format.lower()}"
        )
        
        return f"{self.bucket_name}/{filename}"
    
    def get_image_info(self, filename: str):
        """获取图片信息"""
        stat = self.client.stat_object(self.bucket_name, filename)
        image_data = self.get_image_bytes(filename)
        
        image_stream = io.BytesIO(image_data)
        image = Image.open(image_stream)
        
        return {
            'size': stat.size,
            'modified': stat.last_modified,
            'content_type': stat.content_type,
            'width': image.width,
            'height': image.height,
            'format': image.format
        }
    
    def get_image_bytes(self, filename: str) -> bytes:
        """获取图片字节数据"""
        response = self.client.get_object(self.bucket_name, filename)
        try:
            return response.read()
        finally:
            response.close()
            response.release_conn()
    
    def create_thumbnail(self, original_filename: str, thumbnail_size: Tuple[int, int] = (150, 150)):
        """创建缩略图"""
        # 获取原始图片
        original_data = self.get_image_bytes(original_filename)
        image = Image.open(io.BytesIO(original_data))
        
        # 创建缩略图
        image.thumbnail(thumbnail_size, Image.Resampling.LANCZOS)
        
        # 保存缩略图到字节流
        thumb_stream = io.BytesIO()
        image.save(thumb_stream, format=image.format)
        thumb_stream.seek(0)
        
        # 生成缩略图文件名
        name_parts = original_filename.rsplit('.', 1)
        if len(name_parts) == 2:
            thumbnail_filename = f"{name_parts[0]}_thumb.{name_parts[1]}"
        else:
            thumbnail_filename = f"{original_filename}_thumb"
        
        # 上传缩略图
        self.client.put_object(
            self.bucket_name,
            thumbnail_filename,
            thumb_stream,
            length=thumb_stream.getbuffer().nbytes,
            content_type=f"image/{image.format.lower()}"
        )
        
        return thumbnail_filename

# 使用示例
image_service = ImageStorageService(minio_client)

# 上传图片并创建缩略图
with open("example.jpg", "rb") as f:
    image_data = f.read()
    
original_path = image_service.upload_image(image_data, "example.jpg", resize=(800, 600))
thumbnail_path = image_service.create_thumbnail("example.jpg")

print(f"原始图片: {original_path}")
print(f"缩略图: {thumbnail_path}")

import zipfile
import tarfile
import gzip
from datetime import datetime, timedelta
import shutil

class BackupService:
    def __init__(self, minio_client, backup_bucket="backups"):
        self.client = minio_client
        self.backup_bucket = backup_bucket
        
        if not self.client.bucket_exists(self.backup_bucket):
            self.client.make_bucket(self.backup_bucket)
    
    def backup_directory(self, local_dir_path: str, archive_name: str = None):
        """备份目录到MinIO"""
        if not archive_name:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            archive_name = f"backup_{timestamp}.tar.gz"
        
        # 创建压缩包
        archive_path = f"/tmp/{archive_name}"
        
        with tarfile.open(archive_path, "w:gz") as tar:
            tar.add(local_dir_path, arcname=".")
        
        # 上传到MinIO
        with open(archive_path, "rb") as f:
            self.client.put_object(
                self.backup_bucket,
                archive_name,
                f,
                length=os.path.getsize(archive_path),
                content_type="application/gzip"
            )
        
        # 清理临时文件
        os.remove(archive_path)
        
        return f"{self.backup_bucket}/{archive_name}"
    
    def restore_backup(self, archive_name: str, restore_path: str):
        """从MinIO恢复备份"""
        # 下载备份文件
        backup_data = self.get_backup_bytes(archive_name)
        
        # 创建临时文件
        temp_archive_path = f"/tmp/restore_{archive_name}"
        with open(temp_archive_path, "wb") as f:
            f.write(backup_data)
        
        # 解压到目标路径
        if archive_name.endswith('.tar.gz'):
            with tarfile.open(temp_archive_path, "r:gz") as tar:
                tar.extractall(path=restore_path)
        elif archive_name.endswith('.zip'):
            with zipfile.ZipFile(temp_archive_path, 'r') as zip_ref:
                zip_ref.extractall(restore_path)
        
        # 清理临时文件
        os.remove(temp_archive_path)
        
        return restore_path
    
    def get_backup_bytes(self, archive_name: str) -> bytes:
        """获取备份文件字节数据"""
        response = self.client.get_object(self.backup_bucket, archive_name)
        try:
            return response.read()
        finally:
            response.close()
            response.release_conn()
    
    def list_backups(self):
        """列出所有备份"""
        backups = []
        objects = self.client.list_objects(self.backup_bucket, recursive=True)
        for obj in objects:
            backups.append({
                'name': obj.object_name,
                'size': obj.size,
                'last_modified': obj.last_modified,
                'etag': obj.etag
            })
        return sorted(backups, key=lambda x: x['last_modified'], reverse=True)
    
    def cleanup_old_backups(self, days_to_keep: int = 30):
        """清理旧备份"""
        cutoff_date = datetime.now() - timedelta(days=days_to_keep)
        backups = self.list_backups()
        
        deleted_count = 0
        for backup in backups:
            if backup['last_modified'].replace(tzinfo=None) < cutoff_date:
                self.client.remove_object(self.backup_bucket, backup['name'])
                deleted_count += 1
        
        return deleted_count

# 使用示例
backup_service = BackupService(minio_client)

# 创建备份
backup_path = backup_service.backup_directory("/path/to/data", "daily_backup.tar.gz")
print(f"备份创建成功: {backup_path}")

# 列出备份
backups = backup_service.list_backups()
print(f"共有 {len(backups)} 个备份")

# 清理30天前的备份
deleted = backup_service.cleanup_old_backups(days_to_keep=30)
print(f"清理了 {deleted} 个旧备份")

# 优化MinIO客户端配置
def create_optimized_client():
    client = Minio(
        "localhost:9000",
        access_key="minioadmin",
        secret_key="minioadmin123",
        secure=False,
        # 连接池配置
        http_client=requests.Session(),
    )
    
    # 配置HTTP会话以提高性能
    from requests.adapters import HTTPAdapter
    from urllib3.util.retry import Retry
    
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504],
    )
    adapter = HTTPAdapter(max_retries=retry_strategy, pool_connections=100, pool_maxsize=100)
    
    client.set_http_client(adapter)
    return client

def upload_large_file(client, bucket_name, object_name, file_path, part_size=100*1024*1024):  # 100MB chunks
    """大文件分片上传"""
    from minio.helpers import get_part_info
    
    file_size = os.path.getsize(file_path)
    
    # 开始多部分上传
    upload_id = client._create_multipart_upload(bucket_name, object_name, {})
    
    uploaded_parts = []
    part_number = 1
    bytes_uploaded = 0
    
    with open(file_path, 'rb') as file:
        while bytes_uploaded < file_size:
            length = min(part_size, file_size - bytes_uploaded)
            data = file.read(length)
            
            # 上传分片
            part = client._upload_part(
                bucket_name, object_name, upload_id, part_number, data, len(data), {}
            )
            
            uploaded_parts.append({
                "PartNumber": part_number,
                "ETag": part.etag
            })
            
            bytes_uploaded += length
            part_number += 1
            
            print(f"上传进度: {bytes_uploaded}/{file_size} ({bytes_uploaded/file_size*100:.1f}%)")
    
    # 完成分片上传
    client._complete_multipart_upload(
        bucket_name, object_name, upload_id, uploaded_parts, {}
    )
    
    print(f"大文件上传完成: {object_name}")

def health_check(client):
    """MinIO健康检查"""
    try:
        # 检查连接
        client.list_buckets()
        
        # 检查存储桶
        buckets = client.list_buckets()
        
        # 简单的读写测试
        test_bucket = "health-check-test"
        test_object = "health_test.txt"
        test_data = b"health check data"
        
        # 创建测试存储桶
        if not client.bucket_exists(test_bucket):
            client.make_bucket(test_bucket)
        
        # 上传测试对象
        client.put_object(
            test_bucket,
            test_object,
            test_data,
            length=len(test_data)
        )
        
        # 下载测试对象
        response = client.get_object(test_bucket, test_object)
        downloaded_data = response.read()
        response.close()
        response.release_conn()
        
        # 验证数据完整性
        if downloaded_data == test_data:
            # 清理测试对象
            client.remove_object(test_bucket, test_object)
            print("MinIO健康检查: 通过")
            return True
        else:
            print("MinIO健康检查: 数据不一致")
            return False
            
    except Exception as e:
        print(f"MinIO健康检查: 失败 - {str(e)}")
        return False

"""
MinIO安全最佳实践：
1. 使用强密码：访问密钥和秘密密钥应足够复杂
2. TLS加密：在生产环境中启用HTTPS
3. 最小权限原则：为每个应用创建专用用户和策略
4. 定期轮换密钥：定期更换访问密钥
5. 网络隔离：使用防火墙限制访问
6. 监控审计：启用访问日志记录
"""

# 安全的客户端初始化
def create_secure_client():
    # 使用环境变量存储凭据
    import os
    
    client = Minio(
        os.getenv('MINIO_ENDPOINT', 'localhost:9000'),
        access_key=os.getenv('MINIO_ACCESS_KEY'),
        secret_key=os.getenv('MINIO_SECRET_KEY'),
        secure=os.getenv('MINIO_SECURE', 'true').lower() == 'true'
    )
    return client

"""
MinIO性能最佳实践：
1. 使用SSD存储：获得更好的I/O性能
2. 网络优化：使用高速网络连接
3. 分布式部署：使用多个节点提高可用性和性能
4. 适当分片：大文件使用分片上传
5. 缓存配置：启用适当的缓存策略
6. 对象大小：避免过多小文件，合并小对象
"""

# 批量操作优化
def batch_upload_files(client, bucket_name, file_paths):
    """批量上传文件"""
    import concurrent.futures
    from threading import Lock
    
    upload_results = []
    lock = Lock()
    
    def upload_single_file(file_path):
        filename = os.path.basename(file_path)
        try:
            client.fput_object(bucket_name, filename, file_path)
            with lock:
                upload_results.append({'file': filename, 'status': 'success'})
        except Exception as e:
            with lock:
                upload_results.append({'file': filename, 'status': 'failed', 'error': str(e)})
    
    # 使用线程池并行上传
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = [executor.submit(upload_single_file, fp) for fp in file_paths]
        concurrent.futures.wait(futures)
    
    return upload_results