Module : Data Engineering for AI

# ============================================
# Installation des dépendances
# ============================================

# !pip install boto3 minio pillow pyarrow pandas

import boto3
from botocore.client import Config
import os

# Configuration MinIO
MINIO_ENDPOINT = "http://localhost:9000"
MINIO_ACCESS_KEY = "minioadmin"
MINIO_SECRET_KEY = "minioadmin123"

# Créer le client S3 (compatible MinIO)
s3_client = boto3.client(
    's3',
    endpoint_url=MINIO_ENDPOINT,
    aws_access_key_id=MINIO_ACCESS_KEY,
    aws_secret_access_key=MINIO_SECRET_KEY,
    config=Config(signature_version='s3v4'),
    region_name='us-east-1'  # Requis mais ignoré par MinIO
)

print("✅ Client S3 connecté à MinIO")

# ============================================
# Créer un bucket
# ============================================

BUCKET_NAME = "raw-images"

def create_bucket_if_not_exists(bucket_name: str):
    """Crée un bucket s'il n'existe pas."""
    try:
        s3_client.head_bucket(Bucket=bucket_name)
        print(f"✅ Bucket '{bucket_name}' existe déjà")
    except:
        s3_client.create_bucket(Bucket=bucket_name)
        print(f"✅ Bucket '{bucket_name}' créé")

create_bucket_if_not_exists(BUCKET_NAME)

# Lister les buckets
response = s3_client.list_buckets()
print(f"\n📦 Buckets disponibles :")
for bucket in response['Buckets']:
    print(f"   - {bucket['Name']}")

# ============================================
# Créer des images de test et les uploader
# ============================================

from PIL import Image
import io
import numpy as np
from datetime import datetime

def create_test_image(width: int = 256, height: int = 256) -> bytes:
    """Crée une image de test aléatoire."""
    arr = np.random.randint(0, 255, (height, width, 3), dtype=np.uint8)
    img = Image.fromarray(arr)
    buffer = io.BytesIO()
    img.save(buffer, format='JPEG', quality=85)
    return buffer.getvalue()

def upload_image(bucket: str, key: str, image_bytes: bytes, metadata: dict = None):
    """Upload une image avec ses métadonnées."""
    extra_args = {}
    if metadata:
        extra_args['Metadata'] = {k: str(v) for k, v in metadata.items()}
    
    s3_client.put_object(
        Bucket=bucket,
        Key=key,
        Body=image_bytes,
        ContentType='image/jpeg',
        **extra_args
    )

# Créer et uploader 10 images de test
print("📤 Upload des images de test...")
for i in range(10):
    image_bytes = create_test_image()
    key = f"2024/01/image_{i:04d}.jpg"
    metadata = {
        'width': '256',
        'height': '256',
        'source': 'test',
        'created_at': datetime.now().isoformat()
    }
    upload_image(BUCKET_NAME, key, image_bytes, metadata)
    print(f"   ✅ {key}")

print(f"\n🎉 {10} images uploadées dans '{BUCKET_NAME}'")

# ============================================
# Lister les objets dans un bucket
# ============================================

def list_objects(bucket: str, prefix: str = "") -> list:
    """Liste tous les objets dans un bucket avec un préfixe optionnel."""
    objects = []
    paginator = s3_client.get_paginator('list_objects_v2')
    
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        if 'Contents' in page:
            for obj in page['Contents']:
                objects.append({
                    'key': obj['Key'],
                    'size': obj['Size'],
                    'last_modified': obj['LastModified']
                })
    return objects

# Lister toutes les images
objects = list_objects(BUCKET_NAME, prefix="2024/")
print(f"📋 {len(objects)} objets dans '{BUCKET_NAME}/2024/':\n")
for obj in objects[:5]:  # Afficher les 5 premiers
    print(f"   {obj['key']} ({obj['size']} bytes)")
if len(objects) > 5:
    print(f"   ... et {len(objects) - 5} autres")

# ============================================
# Créer un catalogue Parquet
# ============================================

import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from datetime import datetime
import hashlib

def build_catalog(bucket: str, prefix: str = "") -> pd.DataFrame:
    """Construit un catalogue des images dans un bucket."""
    objects = list_objects(bucket, prefix)
    
    records = []
    for obj in objects:
        # Récupérer les métadonnées de l'objet
        response = s3_client.head_object(Bucket=bucket, Key=obj['key'])
        
        record = {
            'key': obj['key'],
            'bucket': bucket,
            'size_bytes': obj['size'],
            'content_type': response.get('ContentType', 'unknown'),
            'last_modified': obj['last_modified'].isoformat(),
            's3_uri': f"s3://{bucket}/{obj['key']}"
        }
        
        # Ajouter les métadonnées custom si présentes
        if 'Metadata' in response:
            for k, v in response['Metadata'].items():
                record[f'meta_{k}'] = v
        
        records.append(record)
    
    return pd.DataFrame(records)

# Construire le catalogue
catalog_df = build_catalog(BUCKET_NAME, "2024/")
print(f"📊 Catalogue construit : {len(catalog_df)} entrées\n")
print(catalog_df.head())

# ============================================
# Sauvegarder le catalogue en Parquet
# ============================================

# Sauvegarder localement
catalog_df.to_parquet('catalog.parquet', index=False)
print("✅ Catalogue sauvegardé localement: catalog.parquet")

# Uploader dans MinIO
with open('catalog.parquet', 'rb') as f:
    s3_client.put_object(
        Bucket=BUCKET_NAME,
        Key='metadata/catalog.parquet',
        Body=f.read(),
        ContentType='application/octet-stream'
    )
print(f"✅ Catalogue uploadé: s3://{BUCKET_NAME}/metadata/catalog.parquet")

# Vérifier la taille
import os
parquet_size = os.path.getsize('catalog.parquet')
print(f"\n📦 Taille du fichier Parquet: {parquet_size} bytes")

# ============================================
# Créer un shard WebDataset
# ============================================

import tarfile
import json

def create_webdataset_shard(
    bucket: str,
    keys: list,
    output_path: str
) -> int:
    """
    Crée un shard WebDataset à partir d'objets S3.
    
    Returns:
        Nombre d'échantillons dans le shard
    """
    count = 0
    
    with tarfile.open(output_path, 'w') as tar:
        for i, key in enumerate(keys):
            # Télécharger l'image
            response = s3_client.get_object(Bucket=bucket, Key=key)
            image_bytes = response['Body'].read()
            
            # Basename pour WebDataset
            basename = f"{i:06d}"
            
            # Ajouter l'image au tar
            img_info = tarfile.TarInfo(name=f"{basename}.jpg")
            img_info.size = len(image_bytes)
            tar.addfile(img_info, io.BytesIO(image_bytes))
            
            # Ajouter les métadonnées JSON
            metadata = {
                'original_key': key,
                'bucket': bucket,
                'index': i
            }
            json_bytes = json.dumps(metadata).encode('utf-8')
            json_info = tarfile.TarInfo(name=f"{basename}.json")
            json_info.size = len(json_bytes)
            tar.addfile(json_info, io.BytesIO(json_bytes))
            
            count += 1
    
    return count

# Créer un shard avec nos images
objects = list_objects(BUCKET_NAME, "2024/")
keys = [obj['key'] for obj in objects]

count = create_webdataset_shard(BUCKET_NAME, keys, "shard_0000.tar")
print(f"✅ Shard créé: shard_0000.tar ({count} échantillons)")

# Vérifier le contenu
with tarfile.open("shard_0000.tar", 'r') as tar:
    print(f"\n📦 Contenu du shard:")
    for member in tar.getmembers()[:6]:
        print(f"   {member.name} ({member.size} bytes)")

# ============================================
# Analyser le catalogue avec DuckDB
# ============================================

# !pip install duckdb

import duckdb

# Connecter DuckDB et charger le Parquet
con = duckdb.connect()

# Requête directe sur le fichier Parquet
print("📊 Statistiques du catalogue:\n")

# Nombre d'images
result = con.execute("SELECT COUNT(*) as total FROM 'catalog.parquet'").fetchone()
print(f"   Total images: {result[0]}")

# Taille totale
result = con.execute("SELECT SUM(size_bytes) as total_bytes FROM 'catalog.parquet'").fetchone()
print(f"   Taille totale: {result[0] / 1024:.2f} KB")

# Distribution par type
print(f"\n📋 Distribution par content_type:")
result = con.execute("""
    SELECT content_type, COUNT(*) as count 
    FROM 'catalog.parquet' 
    GROUP BY content_type
""").fetchall()
for row in result:
    print(f"   {row[0]}: {row[1]}")

# ============================================
# Requêtes avancées sur le catalogue
# ============================================

# Créer une vue pour faciliter les requêtes
con.execute("""
    CREATE OR REPLACE VIEW images AS 
    SELECT * FROM 'catalog.parquet'
""")

# Trouver les images les plus récentes
print("📅 5 images les plus récentes:")
result = con.execute("""
    SELECT key, size_bytes, last_modified
    FROM images
    ORDER BY last_modified DESC
    LIMIT 5
""").fetchdf()
print(result.to_string(index=False))

# Statistiques de taille
print("\n📏 Statistiques de taille:")
result = con.execute("""
    SELECT 
        MIN(size_bytes) as min_size,
        MAX(size_bytes) as max_size,
        AVG(size_bytes)::INTEGER as avg_size,
        MEDIAN(size_bytes)::INTEGER as median_size
    FROM images
""").fetchdf()
print(result.to_string(index=False))

# ============================================
# Pipeline d'ingestion complet
# ============================================

from dataclasses import dataclass
from typing import Optional, List, Tuple
from PIL import Image
import hashlib

@dataclass
class ImageMetadata:
    """Métadonnées d'une image."""
    key: str
    bucket: str
    size_bytes: int
    width: int
    height: int
    format: str
    mode: str
    md5_hash: str
    s3_uri: str

class IngestionPipeline:
    """Pipeline d'ingestion d'images vers MinIO."""
    
    def __init__(self, s3_client, bucket: str):
        self.s3 = s3_client
        self.bucket = bucket
        self.ingested = []
        self.rejected = []
    
    def validate_image(self, image_bytes: bytes) -> Tuple[bool, Optional[dict], str]:
        """
        Valide une image.
        
        Returns:
            (is_valid, metadata_dict, error_message)
        """
        try:
            img = Image.open(io.BytesIO(image_bytes))
            img.verify()  # Vérifie l'intégrité
            
            # Ré-ouvrir après verify()
            img = Image.open(io.BytesIO(image_bytes))
            
            # Vérifications
            if img.width < 32 or img.height < 32:
                return False, None, "Image trop petite (< 32px)"
            
            if img.width > 10000 or img.height > 10000:
                return False, None, "Image trop grande (> 10000px)"
            
            metadata = {
                'width': img.width,
                'height': img.height,
                'format': img.format or 'UNKNOWN',
                'mode': img.mode
            }
            
            return True, metadata, ""
            
        except Exception as e:
            return False, None, f"Image corrompue: {str(e)}"
    
    def ingest(self, image_bytes: bytes, key: str) -> Optional[ImageMetadata]:
        """
        Ingère une image dans MinIO.
        
        Returns:
            ImageMetadata si succès, None si rejeté
        """
        # Valider
        is_valid, img_meta, error = self.validate_image(image_bytes)
        
        if not is_valid:
            self.rejected.append({'key': key, 'reason': error})
            return None
        
        # Calculer le hash
        md5_hash = hashlib.md5(image_bytes).hexdigest()
        
        # Upload
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=image_bytes,
            ContentType=f"image/{img_meta['format'].lower()}",
            Metadata={
                'width': str(img_meta['width']),
                'height': str(img_meta['height']),
                'md5': md5_hash
            }
        )
        
        # Créer les métadonnées
        metadata = ImageMetadata(
            key=key,
            bucket=self.bucket,
            size_bytes=len(image_bytes),
            width=img_meta['width'],
            height=img_meta['height'],
            format=img_meta['format'],
            mode=img_meta['mode'],
            md5_hash=md5_hash,
            s3_uri=f"s3://{self.bucket}/{key}"
        )
        
        self.ingested.append(metadata)
        return metadata
    
    def get_catalog_df(self) -> pd.DataFrame:
        """Retourne le catalogue sous forme de DataFrame."""
        if not self.ingested:
            return pd.DataFrame()
        return pd.DataFrame([vars(m) for m in self.ingested])
    
    def get_stats(self) -> dict:
        """Retourne les statistiques d'ingestion."""
        return {
            'ingested': len(self.ingested),
            'rejected': len(self.rejected),
            'total': len(self.ingested) + len(self.rejected),
            'success_rate': len(self.ingested) / max(1, len(self.ingested) + len(self.rejected))
        }

print("✅ Pipeline d'ingestion défini")

# ============================================
# Utiliser le pipeline
# ============================================

# Créer un nouveau bucket pour le projet
PROJECT_BUCKET = "project-images"
create_bucket_if_not_exists(PROJECT_BUCKET)

# Initialiser le pipeline
pipeline = IngestionPipeline(s3_client, PROJECT_BUCKET)

# Simuler l'ingestion de 20 images (dont quelques-unes invalides)
print("📤 Ingestion en cours...\n")

for i in range(20):
    # Créer une image (parfois invalide pour tester)
    if i % 7 == 0:  # Simuler une image trop petite
        img = Image.new('RGB', (16, 16), color='red')
    else:
        size = np.random.randint(128, 512)
        arr = np.random.randint(0, 255, (size, size, 3), dtype=np.uint8)
        img = Image.fromarray(arr)
    
    # Convertir en bytes
    buffer = io.BytesIO()
    img.save(buffer, format='JPEG')
    image_bytes = buffer.getvalue()
    
    # Ingérer
    key = f"batch_001/image_{i:04d}.jpg"
    result = pipeline.ingest(image_bytes, key)
    
    status = "✅" if result else "❌"
    print(f"   {status} {key}")

# Afficher les stats
stats = pipeline.get_stats()
print(f"\n📊 Résultats:")
print(f"   ✅ Ingérées: {stats['ingested']}")
print(f"   ❌ Rejetées: {stats['rejected']}")
print(f"   📈 Taux de succès: {stats['success_rate']*100:.1f}%")

# Afficher les rejets
if pipeline.rejected:
    print(f"\n⚠️ Images rejetées:")
    for r in pipeline.rejected:
        print(f"   - {r['key']}: {r['reason']}")

# ============================================
# Sauvegarder le catalogue final
# ============================================

# Obtenir le catalogue
final_catalog = pipeline.get_catalog_df()

if not final_catalog.empty:
    # Sauvegarder en Parquet
    final_catalog.to_parquet('project_catalog.parquet', index=False)
    print("✅ Catalogue sauvegardé: project_catalog.parquet")
    
    # Upload dans MinIO
    with open('project_catalog.parquet', 'rb') as f:
        s3_client.put_object(
            Bucket=PROJECT_BUCKET,
            Key='metadata/catalog.parquet',
            Body=f.read()
        )
    print(f"✅ Catalogue uploadé: s3://{PROJECT_BUCKET}/metadata/catalog.parquet")
    
    # Aperçu
    print(f"\n📊 Aperçu du catalogue:")
    print(final_catalog[['key', 'width', 'height', 'size_bytes']].head())
else:
    print("⚠️ Aucune image ingérée")

# ============================================
# Client multi-cloud abstrait
# ============================================

from enum import Enum
from typing import Optional

class CloudProvider(Enum):
    MINIO = "minio"
    AWS = "aws"
    AZURE = "azure"
    GCP = "gcp"

def create_storage_client(
    provider: CloudProvider,
    endpoint_url: Optional[str] = None,
    access_key: Optional[str] = None,
    secret_key: Optional[str] = None
):
    """
    Crée un client de stockage selon le provider.
    
    En production, utilise des variables d'environnement :
    - AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY
    - AZURE_STORAGE_CONNECTION_STRING
    - GOOGLE_APPLICATION_CREDENTIALS
    """
    
    if provider == CloudProvider.MINIO:
        return boto3.client(
            's3',
            endpoint_url=endpoint_url or 'http://localhost:9000',
            aws_access_key_id=access_key,
            aws_secret_access_key=secret_key,
            config=Config(signature_version='s3v4'),
            region_name='us-east-1'
        )
    
    elif provider == CloudProvider.AWS:
        # Utilise les credentials par défaut (env vars, IAM role, etc.)
        return boto3.client('s3')
    
    elif provider == CloudProvider.GCP:
        # GCS avec interopérabilité S3
        return boto3.client(
            's3',
            endpoint_url='https://storage.googleapis.com',
            aws_access_key_id=access_key,  # GCS HMAC key
            aws_secret_access_key=secret_key
        )
    
    else:
        raise ValueError(f"Provider {provider} non supporté avec boto3")

# Exemple d'utilisation
print("📦 Création de clients multi-cloud:\n")

# MinIO (local)
minio_client = create_storage_client(
    CloudProvider.MINIO,
    endpoint_url=MINIO_ENDPOINT,
    access_key=MINIO_ACCESS_KEY,
    secret_key=MINIO_SECRET_KEY
)
print("   ✅ Client MinIO créé")

# AWS (simulé - échouerait sans credentials)
# aws_client = create_storage_client(CloudProvider.AWS)
print("   ⏸️ Client AWS (nécessite credentials)")

print("\n💡 Le même code fonctionne avec tous les providers !")

# ============================================
# Générer le package téléchargeable
# ============================================

import os
import zipfile

# Créer la structure
os.makedirs('module2-data-storage/src', exist_ok=True)

# requirements.txt
requirements = '''boto3>=1.28.0
minio>=7.1.0
pandas>=2.0.0
pyarrow>=12.0.0
duckdb>=0.8.0
pillow>=9.0.0
'''

with open('module2-data-storage/requirements.txt', 'w') as f:
    f.write(requirements)

# docker-compose.yml
docker_compose = '''version: '3.8'

services:
  minio:
    image: minio/minio:latest
    container_name: minio
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin123
    command: server /data --console-address ":9001"
    volumes:
      - minio_data:/data

volumes:
  minio_data:
'''

with open('module2-data-storage/docker-compose.yml', 'w') as f:
    f.write(docker_compose)

print("✅ Structure du projet créée")
print("\n📦 Pour télécharger, exécute la cellule suivante")

# ============================================
# Créer le ZIP téléchargeable
# ============================================

import zipfile
import base64
from IPython.display import HTML, display

# Créer le ZIP
zip_path = 'module2-data-storage.zip'
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
    for root, dirs, files in os.walk('module2-data-storage'):
        for file in files:
            file_path = os.path.join(root, file)
            arcname = file_path  # Garde la structure
            zipf.write(file_path, arcname)

# Créer le lien de téléchargement
with open(zip_path, 'rb') as f:
    zip_data = f.read()
    b64 = base64.b64encode(zip_data).decode()

# Afficher le bouton de téléchargement
html = f'''
<a download="module2-data-storage.zip" 
   href="data:application/zip;base64,{b64}" 
   style="display: inline-block; padding: 10px 20px; 
          background-color: #4CAF50; color: white; 
          text-decoration: none; border-radius: 5px;
          font-weight: bold;">
   📦 Télécharger module2-data-storage.zip
</a>
'''

display(HTML(html))
print(f"\n✅ Archive créée: {zip_path} ({os.path.getsize(zip_path)} bytes)")