microdao-daarion/scripts/migrate_assets_to_minio.py

#!/usr/bin/env python3
"""
Міграція assets з локального /static/uploads в MinIO
"""
import os
import sys
import mimetypes
from pathlib import Path
from datetime import datetime
from uuid import uuid4

# Додати шлях до city-service для імпорту
sys.path.insert(0, str(Path(__file__).parent.parent / "services" / "city-service"))

try:
    from lib.assets_client import upload_asset, ensure_bucket, client, settings
except ImportError as e:
    print(f"❌ Помилка імпорту: {e}")
    print("Переконайтеся що ви в правильній директорії та MinIO клієнт налаштований")
    sys.exit(1)

def get_content_type(file_path: str) -> str:
    """Визначає content-type файлу"""
    mime_type, _ = mimetypes.guess_type(file_path)
    return mime_type or "application/octet-stream"

def determine_prefix(file_path: str) -> str:
    """Визначає префікс для MinIO на основі шляху файлу"""
    path = Path(file_path)
    name_lower = path.name.lower()
    
    # Логотипи MicroDAO
    if "logo" in name_lower or "microdao" in str(path.parent).lower():
        return "microdao/logo"
    
    # Баннери MicroDAO
    if "banner" in name_lower:
        return "microdao/banner"
    
    # Аватари агентів
    if "avatar" in name_lower or "agent" in name_lower:
        return "agents/avatar"
    
    # Кімнати
    if "room" in name_lower:
        if "logo" in name_lower:
            return "rooms/logo"
        if "banner" in name_lower:
            return "rooms/banner"
    
    # За замовчуванням - misc
    return "misc"

def migrate_file(local_path: str, dry_run: bool = False) -> tuple[str, str] | None:
    """
    Мігрує один файл в MinIO
    Повертає (old_path, new_url) або None якщо помилка
    """
    try:
        content_type = get_content_type(local_path)
        prefix = determine_prefix(local_path)
        
        if dry_run:
            print(f"  [DRY RUN] Would upload: {local_path} -> {prefix}/...")
            return (local_path, f"https://assets.daarion.space/{prefix}/migrated/{Path(local_path).name}")
        
        # Читаємо файл
        with open(local_path, "rb") as f:
            # Завантажуємо в MinIO
            new_url = upload_asset(f, content_type, prefix)
            print(f"  ✅ Uploaded: {local_path} -> {new_url}")
            return (local_path, new_url)
            
    except Exception as e:
        print(f"  ❌ Error migrating {local_path}: {e}")
        return None

def migrate_directory(source_dir: str, dry_run: bool = False) -> dict:
    """
    Мігрує всі файли з директорії
    Повертає статистику: {"total": N, "success": M, "failed": K, "mapping": {...}}
    """
    source_path = Path(source_dir)
    if not source_path.exists():
        print(f"❌ Директорія не існує: {source_dir}")
        return {"total": 0, "success": 0, "failed": 0, "mapping": {}}
    
    # Знаходимо всі файли (виключаємо .git, __pycache__ тощо)
    files = []
    for ext in [".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg", ".ico"]:
        files.extend(source_path.rglob(f"*{ext}"))
        files.extend(source_path.rglob(f"*{ext.upper()}"))
    
    print(f"📦 Знайдено {len(files)} файлів для міграції")
    
    if dry_run:
        print("🔍 DRY RUN MODE - файли не будуть завантажені")
    
    # Переконаємося що bucket існує
    if not dry_run:
        ensure_bucket()
        print(f"✅ Bucket '{settings.assets_bucket}' готовий")
    
    # Мігруємо файли
    stats = {"total": len(files), "success": 0, "failed": 0, "mapping": {}}
    
    for file_path in files:
        result = migrate_file(str(file_path), dry_run)
        if result:
            old_path, new_url = result
            stats["mapping"][old_path] = new_url
            stats["success"] += 1
        else:
            stats["failed"] += 1
    
    return stats

def main():
    import argparse
    
    parser = argparse.ArgumentParser(description="Міграція assets в MinIO")
    parser.add_argument(
        "--source",
        default="/opt/microdao-daarion/services/city-service/static/uploads",
        help="Директорія з assets для міграції"
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Тільки показати що буде зроблено, не завантажувати"
    )
    args = parser.parse_args()
    
    print("🚀 Початок міграції assets в MinIO")
    print(f"📁 Source: {args.source}")
    print(f"🌐 MinIO Endpoint: {settings.minio_endpoint}")
    print(f"🪣 Bucket: {settings.assets_bucket}")
    print(f"🔗 Public URL: {settings.assets_public_base_url}")
    print()
    
    stats = migrate_directory(args.source, dry_run=args.dry_run)
    
    print()
    print("=" * 60)
    print("📊 Статистика міграції:")
    print(f"  Total: {stats['total']}")
    print(f"  ✅ Success: {stats['success']}")
    print(f"  ❌ Failed: {stats['failed']}")
    print("=" * 60)
    
    if not args.dry_run and stats["mapping"]:
        # Зберігаємо маппінг для оновлення БД
        mapping_file = Path(__file__).parent / "assets_migration_mapping.json"
        import json
        with open(mapping_file, "w") as f:
            json.dump(stats["mapping"], f, indent=2)
        print(f"💾 Маппінг збережено в: {mapping_file}")
        print("⚠️  Потрібно оновити URLs в базі даних використовуючи цей маппінг")

if __name__ == "__main__":
    main()