Upload 5 files

Dockerfile (1)

@@ -0,0 +1,52 @@
+FROM python:3.11-slim
+
+# Métadonnées
+LABEL maintainer="dubbing-pipeline"
+LABEL description="Automated video dubbing pipeline - HuggingFace Space"
+
+# Variables d'environnement
+ENV PYTHONUNBUFFERED=1 \
+    PYTHONDONTWRITEBYTECODE=1 \
+    PIP_NO_CACHE_DIR=1 \
+    TRANSFORMERS_CACHE=/app/.cache/huggingface \
+    HF_HOME=/app/.cache/huggingface
+
+# Dépendances système
+# Note : ffmpeg et libsndfile1 sont requis pour le traitement audio/video
+# git-lfs pour les gros modèles HuggingFace
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    ffmpeg \
+    libsndfile1 \
+    libsndfile1-dev \
+    git \
+    git-lfs \
+    wget \
+    curl \
+    espeak-ng \
+    libespeak-ng1 \
+    && apt-get clean \
+    && rm -rf /var/lib/apt/lists/*
+
+# Dossier de travail
+WORKDIR /app
+
+# Créer les dossiers nécessaires
+RUN mkdir -p /app/.cache/huggingface /tmp/dubbing_cache
+
+# Copier le requirements
+COPY requirements.txt .
+
+# Installer les dépendances Python
+# On installe PyTorch CPU en premier (plus léger, Spaces gratuit n'a pas de GPU)
+RUN pip install --upgrade pip && \
+    pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu && \
+    pip install -r requirements.txt
+
+# Copier le code de l'application
+COPY app.py .
+
+# Port Gradio
+EXPOSE 7860
+
+# Commande de démarrage
+CMD ["python", "app.py"]

README.md

@@ -1,11 +1,34 @@
 ---
-title: Tp Project2
-emoji: 🦀
-colorFrom: green
+title: Video Dubbing Pipeline
+emoji: 🎬
+colorFrom: blue
 colorTo: purple
 sdk: docker
 pinned: false
 license: mit
+app_port: 7860
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Automated Video Dubbing Pipeline
+
+Pipeline automatique de doublage vidéo utilisant uniquement des modèles open-weights.
+
+## Comment utiliser
+
+1. Entrez un lien YouTube (vidéo de 30s à 1min)
+2. Choisissez la langue cible
+3. Cliquez sur "Lancer le doublage"
+4. Attendez 5-15 minutes selon la durée
+
+## Stack technique
+
+- **ASR** : OpenAI Whisper (medium)
+- **Traduction** : Qwen2.5-1.5B-Instruct
+- **TTS** : Facebook MMS-TTS
+- **Voice Cloning** : Coqui XTTS-v2 (optionnel)
+- **Mixing** : pydub + ffmpeg
+- **Interface** : Gradio
+
+## Langues supportées
+
+French, Arabic, Spanish, German

app (1).py

@@ -0,0 +1,294 @@
+import os
+import re
+import json
+import datetime
+import subprocess
+import tempfile
+import shutil
+import numpy as np
+import torch
+import srt
+import gradio as gr
+from pathlib import Path
+from pydub import AudioSegment
+from pydub.effects import speedup
+from functools import reduce
+
+import whisper
+from transformers import (
+    AutoTokenizer, AutoModelForCausalLM,
+    VitsModel, AutoTokenizer as TTSTokenizer
+)
+
+# ============================================================
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+
+SUPPORTED_LANGUAGES = {
+    "French": ("facebook/mms-tts-fra", "fr"),
+    "Arabic": ("facebook/mms-tts-ara", "ar"),
+    "Spanish": ("facebook/mms-tts-spa", "es"),
+    "German":  ("facebook/mms-tts-deu", "de"),
+    "English": ("facebook/mms-tts-eng", "en"),
+}
+
+# Cache des modèles (pour éviter de re-télécharger à chaque requête)
+_model_cache = {}
+
+
+def get_whisper():
+    if "whisper" not in _model_cache:
+        _model_cache["whisper"] = whisper.load_model("base", device=DEVICE)
+    return _model_cache["whisper"]
+
+
+def get_llm():
+    mid = "Qwen/Qwen2.5-1.5B-Instruct"
+    if "llm" not in _model_cache:
+        tok = AutoTokenizer.from_pretrained(mid)
+        mdl = AutoModelForCausalLM.from_pretrained(
+            mid, torch_dtype=torch.float16 if DEVICE == "cuda" else torch.float32,
+            device_map="auto"
+        )
+        _model_cache["llm"] = (tok, mdl)
+    return _model_cache["llm"]
+
+
+def get_tts(lang: str):
+    model_id = SUPPORTED_LANGUAGES[lang][0]
+    key = f"tts_{lang}"
+    if key not in _model_cache:
+        tok = TTSTokenizer.from_pretrained(model_id)
+        mdl = VitsModel.from_pretrained(model_id).to(DEVICE)
+        mdl.eval()
+        _model_cache[key] = (tok, mdl)
+    return _model_cache[key]
+
+
+# ---- Pipeline functions ----
+
+def download_video(url: str, work_dir: Path) -> dict:
+    video = work_dir / "video.mp4"
+    audio = work_dir / "audio.wav"
+    # Utilise les cookies si disponibles (contourne le blocage bot YouTube)
+    import os as _os
+    cookies_path = "/app/cookies.txt"
+    yt_cmd = ["yt-dlp", "-f",
+        "bestvideo[height<=480][ext=mp4]+bestaudio[ext=m4a]/best[height<=480][ext=mp4]",
+        "--merge-output-format", "mp4",
+        "-o", str(video), "--no-playlist", url]
+    if _os.path.exists(cookies_path):
+        yt_cmd = ["yt-dlp", "--cookies", cookies_path, "-f",
+            "bestvideo[height<=480][ext=mp4]+bestaudio[ext=m4a]/best[height<=480][ext=mp4]",
+            "--merge-output-format", "mp4",
+            "-o", str(video), "--no-playlist", url]
+    subprocess.run(yt_cmd, check=True, capture_output=True)
+    subprocess.run([
+        "ffmpeg", "-y", "-i", str(video),
+        "-ac", "1", "-ar", "16000", "-vn", str(audio)
+    ], check=True, capture_output=True)
+    probe = subprocess.run([
+        "ffprobe", "-v", "error", "-show_entries", "format=duration",
+        "-of", "json", str(audio)
+    ], capture_output=True, text=True)
+    duration = float(json.loads(probe.stdout)["format"]["duration"])
+    return {"video": video, "audio": audio, "duration": duration}
+
+
+def transcribe(audio_path: Path) -> list:
+    model = get_whisper()
+    result = model.transcribe(str(audio_path), word_timestamps=True, verbose=False)
+    segments = [
+        {"text": s["text"].strip(), "start": round(s["start"], 3),
+         "end": round(s["end"], 3), "duration": round(s["end"] - s["start"], 3)}
+        for s in result["segments"] if s["text"].strip()
+    ]
+    lang = result.get("language", "english").capitalize()
+    return segments, lang
+
+
+def translate_segment(text: str, src: str, tgt: str) -> str:
+    tok, mdl = get_llm()
+    sys_p = (f"You are a professional subtitle translator. Translate from {src} to {tgt}. "
+             f"Output ONLY the translation, nothing else.")
+    msgs = [{"role": "system", "content": sys_p}, {"role": "user", "content": text}]
+    input_text = tok.apply_chat_template(msgs, tokenize=False, add_generation_prompt=True)
+    inputs = tok(input_text, return_tensors="pt").to(DEVICE)
+    with torch.no_grad():
+        out = mdl.generate(**inputs, max_new_tokens=150, temperature=0.3,
+                           do_sample=True, repetition_penalty=1.1,
+                           pad_token_id=tok.eos_token_id)
+    gen = out[0][inputs.input_ids.shape[1]:]
+    tr = tok.decode(gen, skip_special_tokens=True).strip()
+    lines = [l.strip() for l in tr.splitlines() if l.strip()]
+    return lines[0] if lines else tr
+
+
+def build_dubbed_audio(segments: list, tgt_lang: str, total_s: float, out: Path) -> Path:
+    tts_tok, tts_mdl = get_tts(tgt_lang)
+    sr = tts_mdl.config.sampling_rate
+    track = AudioSegment.silent(duration=int(total_s * 1000))
+    for seg in segments:
+        text = seg["translated_text"].strip()
+        if not text:
+            continue
+        inputs = tts_tok(text, return_tensors="pt").to(DEVICE)
+        with torch.no_grad():
+            wav = tts_mdl(**inputs).waveform[0].cpu().numpy()
+        wav_i16 = (wav * 32767).astype(np.int16)
+        seg_aud = AudioSegment(wav_i16.tobytes(), frame_rate=sr, sample_width=2, channels=1)
+        target_s = seg["duration"]
+        actual_s = len(seg_aud) / 1000
+        if actual_s > target_s and target_s > 0.1:
+            spd = min(actual_s / target_s, 2.0)
+            seg_aud = speedup(seg_aud, spd, chunk_size=50, crossfade=25)
+        track = track.overlay(seg_aud, position=int(seg["start"] * 1000))
+    track.export(str(out), format="wav")
+    return out
+
+
+def mix_audio(orig: Path, dubbed: Path, segments: list, out: Path) -> Path:
+    original = AudioSegment.from_wav(str(orig)).set_frame_rate(44100).set_channels(2)
+    dub = AudioSegment.from_wav(str(dubbed)).set_frame_rate(44100).set_channels(2)
+    total = len(original)
+    if len(dub) < total:
+        dub = dub + AudioSegment.silent(duration=total - len(dub), frame_rate=44100)
+    dub = dub[:total]
+    parts, prev = [], 0
+    for seg in segments:
+        s, e = int(seg["start"]*1000), int(seg["end"]*1000)
+        if s > prev:
+            parts.append(original[prev:s])
+        chunk = original[s:e] + (-20)  # ~10% volume
+        parts.append(chunk)
+        prev = e
+    if prev < total:
+        parts.append(original[prev:])
+    ducked = reduce(lambda a, b: a + b, parts) if parts else original
+    final = ducked.overlay(dub + (-0.9))  # dub at ~90%
+    final.export(str(out), format="wav")
+    return out
+
+
+def burn_subtitles(video: Path, audio: Path, srt_path: Path, out: Path) -> Path:
+    srt_esc = str(srt_path).replace("\\", "/")
+    cmd = [
+        "ffmpeg", "-y",
+        "-i", str(video),
+        "-i", str(audio),
+        "-vf", f"subtitles={srt_esc}:force_style='FontSize=18,PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000,Outline=2,Bold=1'",
+        "-c:v", "libx264", "-preset", "ultrafast", "-crf", "28",
+        "-c:a", "aac", "-b:a", "128k",
+        "-map", "0:v:0", "-map", "1:a:0", "-shortest", str(out)
+    ]
+    r = subprocess.run(cmd, capture_output=True, text=True)
+    if r.returncode != 0:
+        # fallback sans sous-titres burned
+        cmd2 = ["ffmpeg", "-y", "-i", str(video), "-i", str(audio),
+                "-c:v", "copy", "-c:a", "aac", "-b:a", "128k",
+                "-map", "0:v:0", "-map", "1:a:0", "-shortest", str(out)]
+        subprocess.run(cmd2, check=True, capture_output=True)
+    return out
+
+
+# ---- PIPELINE COMPLET ----
+
+def run_pipeline(youtube_url: str, target_language: str, progress=gr.Progress()) -> str:
+    """
+    Pipeline principal appelé par Gradio.
+    Retourne le chemin vers la vidéo finale.
+    """
+    if not youtube_url.strip():
+        raise gr.Error("Veuillez entrer un URL YouTube valide")
+
+    work_dir = Path(tempfile.mkdtemp(prefix="dubbing_"))
+
+    try:
+        progress(0.05, desc="Téléchargement de la vidéo...")
+        files = download_video(youtube_url, work_dir)
+
+        progress(0.20, desc="Transcription Whisper...")
+        segments, src_lang = transcribe(files["audio"])
+
+        # SRT original
+        subs_orig = [srt.Subtitle(i+1,
+            datetime.timedelta(seconds=s["start"]),
+            datetime.timedelta(seconds=s["end"]),
+            s["text"]) for i, s in enumerate(segments)]
+
+        progress(0.40, desc="Traduction en cours...")
+        translated = []
+        for seg in segments:
+            tr = translate_segment(seg["text"], src_lang, target_language)
+            translated.append({**seg, "translated_text": tr})
+
+        srt_file = work_dir / "translated.srt"
+        subs_tr = [srt.Subtitle(i+1,
+            datetime.timedelta(seconds=s["start"]),
+            datetime.timedelta(seconds=s["end"]),
+            s["translated_text"]) for i, s in enumerate(translated)]
+        srt_file.write_text(srt.compose(subs_tr), encoding="utf-8")
+
+        progress(0.60, desc="Génération audio (TTS)...")
+        dubbed_wav = work_dir / "dubbed.wav"
+        build_dubbed_audio(translated, target_language, files["duration"], dubbed_wav)
+
+        progress(0.80, desc="Mixage audio...")
+        mixed_wav = work_dir / "mixed.wav"
+        mix_audio(files["audio"], dubbed_wav, translated, mixed_wav)
+
+        progress(0.90, desc="Création vidéo finale...")
+        final_video = work_dir / "final.mp4"
+        burn_subtitles(files["video"], mixed_wav, srt_file, final_video)
+
+        # Copier dans un endroit permanent pour Gradio
+        output_path = Path("/tmp/output_dubbed.mp4")
+        shutil.copy(final_video, output_path)
+
+        progress(1.0, desc="Terminé !")
+        return str(output_path)
+
+    except Exception as e:
+        raise gr.Error(f"Erreur pipeline : {str(e)[:300]}")
+
+    finally:
+        # Nettoyage du dossier temporaire
+        shutil.rmtree(work_dir, ignore_errors=True)
+
+
+# ---- INTERFACE GRADIO ----
+
+with gr.Blocks(title="Video Dubbing Pipeline", theme=gr.themes.Soft()) as demo:
+    gr.Markdown("""
+    # 🎬 Automated Video Dubbing Pipeline
+    Entrez un lien YouTube (30s-1min) et choisissez la langue cible.
+    Le pipeline transcrit, traduit, génère une voix et produit une vidéo doublée.
+
+    > **Note :** Le traitement prend environ 5-15 minutes selon la durée de la vidéo.
+    """)
+
+    with gr.Row():
+        with gr.Column(scale=2):
+            url_input = gr.Textbox(
+                label="YouTube URL",
+                placeholder="https://www.youtube.com/watch?v=...",
+                lines=1
+            )
+            lang_choice = gr.Dropdown(
+                choices=list(SUPPORTED_LANGUAGES.keys()),
+                value="French",
+                label="Langue cible"
+            )
+            run_btn = gr.Button("🚀 Lancer le doublage", variant="primary")
+
+        with gr.Column(scale=3):
+            video_output = gr.Video(label="Vidéo doublée")
+
+    run_btn.click(
+        fn=run_pipeline,
+        inputs=[url_input, lang_choice],
+        outputs=video_output
+    )
+
+
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860)

requirements (1).txt

@@ -0,0 +1,32 @@
+# Core ML
+# Note: torch est installé séparément dans le Dockerfile (version CPU)
+transformers>=4.40.0
+accelerate>=0.27.0
+
+# ASR
+openai-whisper>=20231117
+
+# TTS
+# TTS — MMS-TTS est inclus dans transformers, pas d'install séparée
+# Voice cloning via kokoro (compatible Python 3.12+)
+kokoro>=0.9.4
+misaki[en]
+
+# Audio
+pydub>=0.25.1
+soundfile>=0.12.1
+librosa>=0.10.0
+scipy>=1.11.0
+
+# Vidéo / subtitles
+srt>=3.5.3
+
+# Téléchargement YouTube
+yt-dlp>=2024.1.0
+
+# Interface
+gradio>=4.20.0
+
+# Utils
+numpy>=1.24.0
+tqdm>=4.65.0