app.py

# -*- coding: utf-8 -*-
"""
Application de traduction automatique Français -> Mooré (batch CSV/Excel)
Déployable sur Hugging Face Spaces (SDK: Gradio)
 
Fonctionnalités :
  - Upload d'un fichier CSV / Excel contenant une colonne "français"
  - Choix d'un modèle de traduction (Masakhane M2M100 ou NLLB-200)
  - Traduction ligne par ligne avec barre de progression
  - Export du résultat en CSV (utf-8-sig) ou Excel
"""
 
import os
import traceback
 
# IMPORTANT : sur un Space Hugging Face utilisant le matériel "ZeroGPU",
# le paquet `spaces` DOIT être importé avant tout paquet lié à CUDA
# (notamment avant `torch`), sous peine d'erreur au démarrage. En local
# (ou sur un Space CPU/GPU classique), ce paquet n'existe pas : on gère
# donc cet import de façon optionnelle, mais toujours en premier.
try:
    import spaces  # noqa: E402  (doit rester avant l'import de torch)
    _SPACES_DISPONIBLE = True
except ImportError:
    _SPACES_DISPONIBLE = False
 
import gradio as gr
import pandas as pd
import torch
from transformers import AutoModelForSeq2SeqLM, AutoTokenizer
 
# ---------------------------------------------------------------------------
# 1. CONFIGURATION DES MODÈLES DISPONIBLES
# ---------------------------------------------------------------------------
# Chaque entrée décrit comment charger le modèle et comment forcer la langue
# cible lors de la génération. Les identifiants correspondent aux dépôts
# Hugging Face réels (les noms donnés par l'utilisateur ont été normalisés).
MODELES = {
    "Masakhane M2M100 (fr -> mos, spécialisé actualités)": {
        "repo_id": "masakhane/m2m100_418M_fr_mos_news",
        "type": "m2m100",
        "src_lang": "fr",
        "tgt_lang": "mos",
    },
    "NLLB-200 3.3B (Meta, multilingue)": {
        "repo_id": "facebook/nllb-200-3.3B",
        "type": "nllb",
        "src_lang": "fra_Latn",
        "tgt_lang": "mos_Latn",
    },
    "NLLB-200 distilled 600M (Meta, plus rapide)": {
        "repo_id": "facebook/nllb-200-distilled-600M",
        "type": "nllb",
        "src_lang": "fra_Latn",
        "tgt_lang": "mos_Latn",
    },
}
 
# Cache (singleton) : évite de recharger un modèle déjà utilisé
_CACHE_MODELES = {}
 
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
 
 
# ---------------------------------------------------------------------------
# 2. CHARGEMENT DU MODÈLE (mise en cache / singleton)
# ---------------------------------------------------------------------------
def load_model(nom_modele: str):
    """
    Charge (ou récupère depuis le cache) le tokenizer et le modèle
    correspondant au nom sélectionné dans le menu déroulant.
    Déplace automatiquement le modèle sur le GPU si disponible.
    """
    if nom_modele in _CACHE_MODELES:
        return _CACHE_MODELES[nom_modele]
 
    config = MODELES[nom_modele]
    repo_id = config["repo_id"]
 
    tokenizer = AutoTokenizer.from_pretrained(repo_id)
    modele = AutoModelForSeq2SeqLM.from_pretrained(repo_id)
    modele.to(DEVICE)
    modele.eval()
 
    _CACHE_MODELES[nom_modele] = (tokenizer, modele, config)
    return tokenizer, modele, config
 
 
# ---------------------------------------------------------------------------
# 3. TRADUCTION D'UN TEXTE
# ---------------------------------------------------------------------------
def translate_text(texte: str, tokenizer, modele, config) -> str:
    """
    Traduit une seule chaîne de caractères du français vers le mooré.
    Renvoie une chaîne vide si le texte d'entrée est vide/NaN.
    """
    if texte is None:
        return ""
    texte = str(texte).strip()
    if texte == "" or texte.lower() == "nan":
        return ""
 
    type_modele = config["type"]
 
    with torch.no_grad():
        if type_modele == "nllb":
            # Les modèles NLLB nécessitent la langue source sur le tokenizer
            # et l'identifiant de la langue cible comme premier token forcé.
            tokenizer.src_lang = config["src_lang"]
            entrees = tokenizer(texte, return_tensors="pt", truncation=True, max_length=512).to(DEVICE)
 
            # Compatibilité selon les versions de transformers
            try:
                tgt_id = tokenizer.convert_tokens_to_ids(config["tgt_lang"])
            except Exception:
                tgt_id = tokenizer.lang_code_to_id[config["tgt_lang"]]
 
            sortie = modele.generate(
                **entrees,
                forced_bos_token_id=tgt_id,
                max_length=512,
                num_beams=4,
            )
        else:  # "m2m100" (y compris le modèle Masakhane finetuné)
            tokenizer.src_lang = config["src_lang"]
            entrees = tokenizer(texte, return_tensors="pt", truncation=True, max_length=512).to(DEVICE)
 
            try:
                tgt_id = tokenizer.get_lang_id(config["tgt_lang"])
            except Exception:
                # Le tokenizer finetuné n'a parfois pas cette méthode : on
                # se rabat sur une génération sans forçage de langue cible.
                tgt_id = None
 
            if tgt_id is not None:
                sortie = modele.generate(**entrees, forced_bos_token_id=tgt_id, max_length=512, num_beams=4)
            else:
                sortie = modele.generate(**entrees, max_length=512, num_beams=4)
 
    traduction = tokenizer.batch_decode(sortie, skip_special_tokens=True)[0]
    return traduction.strip()
 
 
# ---------------------------------------------------------------------------
# 4. LECTURE DU FICHIER D'ENTRÉE
# ---------------------------------------------------------------------------
def lire_fichier(chemin_fichier: str) -> pd.DataFrame:
    """
    Lit un fichier CSV ou Excel et renvoie un DataFrame.
    Lève une exception explicite en cas de problème de lecture.
    """
    extension = os.path.splitext(chemin_fichier)[1].lower()
 
    try:
        if extension == ".csv":
            # On essaie plusieurs encodages courants pour plus de robustesse
            try:
                df = pd.read_csv(chemin_fichier, encoding="utf-8-sig")
            except UnicodeDecodeError:
                df = pd.read_csv(chemin_fichier, encoding="latin-1")
        elif extension in (".xlsx", ".xls"):
            df = pd.read_excel(chemin_fichier)
        else:
            raise ValueError(f"Format de fichier non supporté : {extension}")
    except Exception as e:
        raise ValueError(f"Impossible de lire le fichier ({e})")
 
    return df
 
 
# ---------------------------------------------------------------------------
# 5. TRAITEMENT COMPLET DU FICHIER (fonction appelée par le bouton Gradio)
# ---------------------------------------------------------------------------
def _process_file_impl(fichier, nom_modele, format_sortie, progress=gr.Progress()):
    """
    Fonction principale déclenchée par le bouton "Lancer la traduction".
    - Lit le fichier
    - Traduit chaque ligne de la colonne "français"
    - Écrit/écrase la colonne "moore"
    - Exporte le résultat au format choisi (CSV ou Excel)
    """
    if fichier is None:
        return None, "❌ Aucun fichier n'a été fourni."
 
    # --- Lecture du fichier ---
    try:
        df = lire_fichier(fichier.name)
    except Exception as e:
        return None, f"❌ Erreur de lecture du fichier : {e}"
 
    if "français" not in df.columns:
        return None, "❌ Le fichier doit contenir une colonne nommée exactement 'français'."
 
    # --- Chargement du modèle (mis en cache après le premier appel) ---
    progress(0, desc="Chargement du modèle...")
    try:
        tokenizer, modele, config = load_model(nom_modele)
    except Exception as e:
        return None, f"❌ Impossible de charger le modèle sélectionné : {e}"
 
    # --- Traduction ligne par ligne avec barre de progression ---
    nb_lignes = len(df)
    traductions = []
    nb_reussies = 0
    nb_vides = 0
    nb_erreurs = 0
 
    for i, valeur in enumerate(df["français"].tolist()):
        progress((i + 1) / max(nb_lignes, 1), desc=f"Traduction {i + 1}/{nb_lignes}")
 
        texte = "" if pd.isna(valeur) else str(valeur)
 
        if texte.strip() == "":
            traductions.append("")
            nb_vides += 1
            continue
 
        try:
            resultat = translate_text(texte, tokenizer, modele, config)
            traductions.append(resultat)
            nb_reussies += 1
        except Exception:
            # On n'interrompt jamais le traitement global : on trace
            # l'erreur dans la cellule et on continue avec la ligne suivante.
            traductions.append("[ERREUR]")
            nb_erreurs += 1
            print(f"[ERREUR ligne {i}] {traceback.format_exc()}")
 
    # La colonne "moore" est créée ou écrasée si elle existait déjà
    df["moore"] = traductions
 
    # --- Export du fichier résultat ---
    nom_base = "resultat_traduction"
    dossier_sortie = "sorties_traduction"
    os.makedirs(dossier_sortie, exist_ok=True)
 
    if format_sortie == "Excel (.xlsx)":
        chemin_sortie = os.path.join(dossier_sortie, f"{nom_base}.xlsx")
        df.to_excel(chemin_sortie, index=False)
    else:  # "CSV (.csv)"
        chemin_sortie = os.path.join(dossier_sortie, f"{nom_base}.csv")
        # Encodage UTF-8 avec BOM pour une ouverture correcte dans Excel/Windows
        df.to_csv(chemin_sortie, index=False, encoding="utf-8-sig")
 
    rapport = (
        f"✅ Traitement terminé sur {DEVICE.upper()}.\n"
        f"Total : {nb_lignes} lignes | "
        f"Traduites : {nb_reussies} | "
        f"Vides ignorées : {nb_vides} | "
        f"Erreurs : {nb_erreurs}"
    )
 
    return chemin_sortie, rapport
 
 
# Sur un Space "ZeroGPU", la fonction qui utilise réellement le GPU doit être
# décorée avec @spaces.GPU pour que Hugging Face lui alloue dynamiquement une
# carte le temps de l'exécution. `duration` (en secondes) borne le temps
# d'allocation GPU accordé pour un appel : on le majore pour laisser le temps
# de traduire des fichiers de plusieurs centaines de lignes.
if _SPACES_DISPONIBLE:
    process_file = spaces.GPU(duration=120)(_process_file_impl)
else:
    process_file = _process_file_impl
 
 
# ---------------------------------------------------------------------------
# 6. INTERFACE GRADIO
# ---------------------------------------------------------------------------
with gr.Blocks(title="Traducteur Français ↔ Mooré") as demo:
    gr.Markdown(
        """
        # 🇧🇫 Traducteur Français → Mooré (traitement par lots)
        Téléversez un fichier CSV ou Excel contenant une colonne **"français"**,
        choisissez un modèle de traduction, puis lancez le traitement.
        Le fichier résultat contiendra une colonne **"moore"** avec les traductions.
        """
    )
 
    with gr.Row():
        with gr.Column():
            fichier_entree = gr.File(
                label="Fichier à traduire (.csv, .xlsx, .xls)",
                file_types=[".csv", ".xlsx", ".xls"],
            )
            choix_modele = gr.Dropdown(
                label="Modèle de traduction",
                choices=list(MODELES.keys()),
                value=list(MODELES.keys())[0],
            )
            format_sortie = gr.Radio(
                label="Format du fichier de sortie",
                choices=["CSV (.csv)", "Excel (.xlsx)"],
                value="CSV (.csv)",
            )
            bouton_lancer = gr.Button("🚀 Lancer la traduction", variant="primary")
 
        with gr.Column():
            fichier_sortie = gr.File(label="Fichier traduit (téléchargement)")
            statut = gr.Textbox(label="Statut / Rapport", lines=4, interactive=False)
 
    bouton_lancer.click(
        fn=process_file,
        inputs=[fichier_entree, choix_modele, format_sortie],
        outputs=[fichier_sortie, statut],
    )
 
    gr.Markdown(
        f"""
        ---
        ℹ️ Appareil de calcul détecté : **{DEVICE.upper()}**.
        Les modèles sont chargés une seule fois puis conservés en mémoire (cache).
        """
    )
 
if __name__ == "__main__":
    demo.launch()